X-Git-Url: https://git.sur5r.net/?a=blobdiff_plain;f=libraries%2Fliblmdb%2Fmdb.c;h=c98247c2b9937afd26a15fa592a53a9ec14d5bbc;hb=7fe85f5c8291f531f93c28b98865367aeca75484;hp=e52da918a43c563c1395d6fdef7ae7f4efde2ea4;hpb=dddc7a385ba4951c11e1c8405ccf002b6a75a53d;p=openldap diff --git a/libraries/liblmdb/mdb.c b/libraries/liblmdb/mdb.c index e52da918a4..c98247c2b9 100644 --- a/libraries/liblmdb/mdb.c +++ b/libraries/liblmdb/mdb.c @@ -1,11 +1,11 @@ /** @file mdb.c - * @brief memory-mapped database library + * @brief Lightning memory-mapped database library * * A Btree-based database management library modeled loosely on the * 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 @@ -37,10 +37,26 @@ #endif #include #include -#include #ifdef _WIN32 #include +/** 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 #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 #include #include #ifdef HAVE_SYS_FILE_H @@ -49,7 +65,6 @@ #include #endif -#include #include #include #include @@ -130,16 +145,27 @@ # error "Two's complement, reasonably sized integer types, please" #endif -/** @defgroup internal MDB Internals +/** @defgroup internal LMDB 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 (), update the #if statements above */ +# define mdb_func_ "" +#endif + #ifdef _WIN32 #define MDB_USE_HASH 1 #define MDB_PIDLOCK 0 @@ -311,7 +337,7 @@ static txnid_t mdb_debug_start; */ # 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 @@ -324,10 +350,13 @@ static txnid_t mdb_debug_start; (((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 @@ -336,7 +365,7 @@ static txnid_t mdb_debug_start; * 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 @@ -352,7 +381,7 @@ static txnid_t mdb_debug_start; */ #define MDB_MINKEYS 2 - /** A stamp that identifies a file as an MDB file. + /** A stamp that identifies a file as an LMDB file. * There's nothing special about this value other than that it is easily * recognizable, and it will reflect any byte order mismatches. */ @@ -363,20 +392,25 @@ static txnid_t mdb_debug_start; /** The version number for a database's lockfile format. */ #define MDB_LOCK_VERSION 1 - /** @brief The maximum size of a key in the database. - * - * The library rejects bigger keys, and cannot deal with records - * with bigger keys stored by a library with bigger max keysize. + /** @brief The max size of a key we can write, or 0 for dynamic 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. + * 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. * - * 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. @@ -386,11 +420,15 @@ static txnid_t mdb_debug_start; #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. @@ -409,6 +447,9 @@ static txnid_t mdb_debug_start; /** 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. @@ -494,7 +535,7 @@ typedef struct MDB_rxbody { */ 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; @@ -527,7 +568,7 @@ typedef struct MDB_reader { * unlikely. If a collision occurs, the results are unpredictable. */ typedef struct MDB_txbody { - /** Stamp identifying this as an MDB file. It must be set + /** Stamp identifying this as an LMDB file. It must be set * to #MDB_MAGIC. */ uint32_t mtb_magic; /** Format of this lock file. Must be set to #MDB_LOCK_FORMAT. */ @@ -609,6 +650,7 @@ typedef struct MDB_page { #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 */ @@ -660,7 +702,8 @@ typedef struct 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 @@ -669,9 +712,11 @@ typedef struct MDB_node { * 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. @@ -797,7 +842,7 @@ typedef struct MDB_db { * Pages 0-1 are meta pages. Transaction N writes meta page #(N % 2). */ typedef struct MDB_meta { - /** Stamp identifying this as an MDB file. It must be set + /** Stamp identifying this as an LMDB file. It must be set * to #MDB_MAGIC. */ uint32_t mm_magic; /** Version number of this lock file. Must be set to #MDB_DATA_VERSION. */ @@ -813,19 +858,18 @@ typedef struct MDB_meta { 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 @@ -893,12 +937,12 @@ struct MDB_txn { * @{ */ #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. @@ -978,6 +1022,7 @@ typedef struct MDB_xcursor { 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. */ @@ -991,19 +1036,19 @@ struct MDB_env { #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 */ @@ -1014,6 +1059,7 @@ struct MDB_env { 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; @@ -1023,6 +1069,10 @@ struct MDB_env { 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 */ @@ -1031,6 +1081,8 @@ struct MDB_env { 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 */ @@ -1046,9 +1098,13 @@ typedef struct MDB_ntxn { #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); @@ -1079,7 +1135,7 @@ static void mdb_env_close0(MDB_env *env, int excl); 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); @@ -1092,7 +1148,8 @@ static int mdb_update_key(MDB_cursor *mc, MDB_val *key); 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); @@ -1128,7 +1185,7 @@ mdb_version(int *major, int *minor, int *patch) return MDB_VERSION_STRING; } -/** Table of descriptions for MDB @ref errors */ +/** Table of descriptions for LMDB @ref errors */ static char *const mdb_errstr[] = { "MDB_KEYEXIST: Key/data pair already exists", "MDB_NOTFOUND: No matching key/data pair found", @@ -1136,7 +1193,7 @@ static char *const mdb_errstr[] = { "MDB_CORRUPTED: Located page was wrong type", "MDB_PANIC: Update of meta page failed", "MDB_VERSION_MISMATCH: Database environment version mismatch", - "MDB_INVALID: File is not an MDB file", + "MDB_INVALID: File is not an LMDB file", "MDB_MAP_FULL: Environment mapsize limit reached", "MDB_DBS_FULL: Environment maxdbs limit reached", "MDB_READERS_FULL: Environment maxreaders limit reached", @@ -1148,7 +1205,7 @@ static char *const mdb_errstr[] = { "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 * @@ -1166,7 +1223,43 @@ mdb_strerror(int err) 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. @@ -1182,7 +1275,7 @@ mdb_dkey(MDB_val *key, char *buf) 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. @@ -1197,33 +1290,77 @@ mdb_dkey(MDB_val *key, char *buf) 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; imp_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 @@ -1249,6 +1386,7 @@ mdb_cursor_chk(MDB_cursor *mc) /** 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) { @@ -1262,10 +1400,13 @@ 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; imt_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; @@ -1273,8 +1414,8 @@ static void mdb_audit(MDB_txn *txn) 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]; @@ -1288,7 +1429,7 @@ static void mdb_audit(MDB_txn *txn) } } } - while (mdb_cursor_sibling(&mc, 1) == 0); + mdb_tassert(txn, rc == MDB_NOTFOUND); } } if (freecount + count + 2 /* metapages */ != txn->mt_next_pgno) { @@ -1318,7 +1459,12 @@ mdb_page_malloc(MDB_txn *txn, unsigned num) { 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); @@ -1326,15 +1472,22 @@ mdb_page_malloc(MDB_txn *txn, unsigned num) 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.) @@ -1374,6 +1527,23 @@ mdb_dlist_free(MDB_txn *txn) 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: @@ -1422,6 +1592,12 @@ mdb_pages_xkeep(MDB_cursor *mc, unsigned pflags, int all) 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; imt_numdbs; i++) { @@ -1596,7 +1772,7 @@ static void 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; @@ -1605,7 +1781,8 @@ mdb_page_dirty(MDB_txn *txn, MDB_page *mp) } 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--; } @@ -1638,21 +1815,34 @@ mdb_page_alloc(MDB_cursor *mc, int num, MDB_page **mp) #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; @@ -1662,14 +1852,14 @@ mdb_page_alloc(MDB_cursor *mc, int num, MDB_page **mp) /* 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; } @@ -1697,7 +1887,7 @@ mdb_page_alloc(MDB_cursor *mc, int num, MDB_page **mp) if (rc) { if (rc == MDB_NOTFOUND) break; - return rc; + goto fail; } last = *(txnid_t*)key.mv_data; if (oldest <= last) @@ -1710,11 +1900,13 @@ mdb_page_alloc(MDB_cursor *mc, int num, MDB_page **mp) 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; @@ -1743,15 +1935,18 @@ mdb_page_alloc(MDB_cursor *mc, int num, MDB_page **mp) 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; @@ -1766,6 +1961,10 @@ search_done: *mp = np; return MDB_SUCCESS; + +fail: + txn->mt_flags |= MDB_TXN_ERROR; + return rc; } /** Copy the used portions of a non-overflow page. @@ -1796,7 +1995,7 @@ mdb_page_copy(MDB_page *dst, MDB_page *src, unsigned int psize) * 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. */ @@ -1872,17 +2071,17 @@ mdb_page_touch(MDB_cursor *mc) 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) { @@ -1903,19 +2102,21 @@ mdb_page_touch(MDB_cursor *mc) 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; } @@ -1941,6 +2142,7 @@ done: 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]); @@ -1951,6 +2153,10 @@ done: } } return 0; + +fail: + txn->mt_flags |= MDB_TXN_ERROR; + return rc; } int @@ -2078,7 +2284,7 @@ enum Pidlock_op { * 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; @@ -2143,16 +2349,14 @@ mdb_txn_renew0(MDB_txn *txn) 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); @@ -2482,7 +2686,7 @@ mdb_freelist_save(MDB_txn *txn) 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); @@ -2493,9 +2697,47 @@ mdb_freelist_save(MDB_txn *txn) 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. @@ -2506,7 +2748,7 @@ mdb_freelist_save(MDB_txn *txn) 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; @@ -2579,7 +2821,12 @@ mdb_freelist_save(MDB_txn *txn) 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; } @@ -2591,22 +2838,20 @@ mdb_freelist_save(MDB_txn *txn) 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; @@ -2762,8 +3007,8 @@ mdb_txn_commit(MDB_txn *txn) 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); @@ -2871,7 +3116,7 @@ mdb_txn_commit(MDB_txn *txn) 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; @@ -2970,10 +3215,11 @@ fail: 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. @@ -2985,13 +3231,13 @@ mdb_env_read_header(MDB_env *env, MDB_meta *meta) 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; @@ -3103,9 +3349,6 @@ mdb_env_write_meta(MDB_txn *txn) 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)); @@ -3122,11 +3365,18 @@ mdb_env_write_meta(MDB_txn *txn) 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; } @@ -3232,6 +3482,7 @@ mdb_env_create(MDB_env **env) 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; @@ -3372,7 +3623,7 @@ mdb_env_get_maxreaders(MDB_env *env, unsigned int *readers) return MDB_SUCCESS; } -/** Further setup required for opening an MDB environment +/** Further setup required for opening an LMDB environment */ static int mdb_env_open2(MDB_env *env) @@ -3397,7 +3648,9 @@ mdb_env_open2(MDB_env *env) 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; } @@ -3417,7 +3670,7 @@ mdb_env_open2(MDB_env *env) 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; @@ -3429,10 +3682,15 @@ mdb_env_open2(MDB_env *env) 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); @@ -3487,7 +3745,9 @@ static void NTAPI mdb_tls_callback(PVOID module, DWORD reason, PVOID ptr) case DLL_THREAD_DETACH: for (i=0; ime_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 @@ -3928,8 +4188,13 @@ fail: * 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) +#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) @@ -4042,7 +4307,12 @@ 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: @@ -4066,6 +4336,7 @@ mdb_env_close0(MDB_env *env, int excl) 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); @@ -4093,7 +4364,7 @@ mdb_env_close0(MDB_env *env, int excl) 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. */ @@ -4339,10 +4610,26 @@ mdb_cmp_cint(const MDB_val *a, const MDB_val *b) } while(!x && u > (unsigned short *)a->mv_data); return x; #else - return memcmp(a->mv_data, b->mv_data, a->mv_size); + unsigned short *u, *c, *end; + int x; + + end = (unsigned short *) ((char *) a->mv_data + a->mv_size); + u = (unsigned short *)a->mv_data; + c = (unsigned short *)b->mv_data; + do { + x = *u++ - *c++; + } while(!x && u < end); + return x; #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) @@ -4409,17 +4696,9 @@ mdb_node_search(MDB_cursor *mc, MDB_val *key, int *exactp) 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; @@ -4483,7 +4762,7 @@ mdb_node_search(MDB_cursor *mc, MDB_val *key, int *exactp) 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) @@ -4533,7 +4812,7 @@ mdb_cursor_push(MDB_cursor *mc, MDB_page *mp) 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; } @@ -4593,7 +4872,7 @@ mdb_page_get(MDB_txn *txn, pgno_t pgno, MDB_page **ret, int *lvl) 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; } @@ -4619,7 +4898,7 @@ mdb_page_search_root(MDB_cursor *mc, MDB_val *key, int flags) 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)) { @@ -4634,14 +4913,14 @@ mdb_page_search_root(MDB_cursor *mc, MDB_val *key, int flags) 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) @@ -4661,6 +4940,7 @@ mdb_page_search_root(MDB_cursor *mc, MDB_val *key, int flags) 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; } @@ -4755,7 +5035,7 @@ mdb_page_search(MDB_cursor *mc, MDB_val *key, int flags) } } - 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; @@ -4825,7 +5105,7 @@ mdb_ovpage_free(MDB_cursor *mc, MDB_page *mp) 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; @@ -4893,20 +5173,14 @@ mdb_get(MDB_txn *txn, MDB_dbi dbi, 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); } @@ -4952,7 +5226,7 @@ mdb_cursor_sibling(MDB_cursor *mc, int move_right) 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) { @@ -4980,7 +5254,7 @@ mdb_cursor_next(MDB_cursor *mc, MDB_val *key, MDB_val *data, MDB_cursor_op op) return MDB_NOTFOUND; } - assert(mc->mc_flags & C_INITIALIZED); + mdb_cassert(mc, mc->mc_flags & C_INITIALIZED); mp = mc->mc_pg[mc->mc_top]; @@ -5002,7 +5276,8 @@ mdb_cursor_next(MDB_cursor *mc, MDB_val *key, MDB_val *data, MDB_cursor_op op) } } - 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; @@ -5019,7 +5294,7 @@ mdb_cursor_next(MDB_cursor *mc, MDB_val *key, MDB_val *data, MDB_cursor_op op) 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; @@ -5027,7 +5302,7 @@ skip: 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)) { @@ -5056,7 +5331,7 @@ mdb_cursor_prev(MDB_cursor *mc, MDB_val *key, MDB_val *data, MDB_cursor_op op) 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]; @@ -5078,7 +5353,8 @@ mdb_cursor_prev(MDB_cursor *mc, MDB_val *key, MDB_val *data, MDB_cursor_op op) } } - 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"); @@ -5094,7 +5370,7 @@ mdb_cursor_prev(MDB_cursor *mc, MDB_val *key, MDB_val *data, MDB_cursor_op op) 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; @@ -5102,7 +5378,7 @@ mdb_cursor_prev(MDB_cursor *mc, MDB_val *key, MDB_val *data, MDB_cursor_op op) 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)) { @@ -5133,8 +5409,6 @@ mdb_cursor_set(MDB_cursor *mc, MDB_val *key, MDB_val *data, MDB_node *leaf = NULL; DKBUF; - assert(mc); - assert(key); if (key->mv_size == 0) return MDB_BAD_VALSIZE; @@ -5224,7 +5498,7 @@ mdb_cursor_set(MDB_cursor *mc, MDB_val *key, MDB_val *data, 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 @@ -5237,7 +5511,7 @@ mdb_cursor_set(MDB_cursor *mc, MDB_val *key, MDB_val *data, 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); @@ -5251,7 +5525,7 @@ set2: 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); } @@ -5260,8 +5534,10 @@ set1: 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; } @@ -5293,6 +5569,7 @@ set1: if (op == MDB_GET_BOTH || rc > 0) return MDB_NOTFOUND; rc = 0; + *data = d2; } } else { @@ -5326,7 +5603,7 @@ mdb_cursor_first(MDB_cursor *mc, MDB_val *key, MDB_val *data) 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; @@ -5372,7 +5649,7 @@ mdb_cursor_last(MDB_cursor *mc, MDB_val *key, MDB_val *data) 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; @@ -5409,7 +5686,8 @@ mdb_cursor_get(MDB_cursor *mc, MDB_val *key, MDB_val *data, 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; @@ -5420,8 +5698,9 @@ mdb_cursor_get(MDB_cursor *mc, MDB_val *key, MDB_val *data, rc = EINVAL; } else { MDB_page *mp = mc->mc_pg[mc->mc_top]; - if (!NUMKEYS(mp)) { - mc->mc_ki[mc->mc_top] = 0; + int nkeys = NUMKEYS(mp); + if (!nkeys || mc->mc_ki[mc->mc_top] >= nkeys) { + mc->mc_ki[mc->mc_top] = nkeys; rc = MDB_NOTFOUND; break; } @@ -5460,12 +5739,10 @@ mdb_cursor_get(MDB_cursor *mc, MDB_val *key, MDB_val *data, 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)) { @@ -5602,19 +5879,24 @@ mdb_cursor_put(MDB_cursor *mc, MDB_val *key, MDB_val *data, 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. */ @@ -5631,14 +5913,14 @@ mdb_cursor_put(MDB_cursor *mc, MDB_val *key, MDB_val *data, 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 @@ -5722,164 +6004,172 @@ mdb_cursor_put(MDB_cursor *mc, MDB_val *key, MDB_val *data, 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; imp_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; @@ -5887,7 +6177,7 @@ current: /* 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) @@ -5902,14 +6192,15 @@ current: */ 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. @@ -5927,46 +6218,42 @@ current: 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; @@ -5986,9 +6273,7 @@ new_sub: } } - 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 @@ -5996,6 +6281,7 @@ new_sub: */ if (do_sub) { int xflags; + size_t ecount; put_sub: xdata.mv_size = 0; xdata.mv_data = ""; @@ -6011,7 +6297,7 @@ put_sub: 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; @@ -6029,6 +6315,7 @@ put_sub: /* 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); @@ -6036,32 +6323,38 @@ put_sub: 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; } @@ -6069,6 +6362,7 @@ int mdb_cursor_del(MDB_cursor *mc, unsigned int flags) { MDB_node *leaf; + MDB_page *mp; int rc; if (mc->mc_txn->mt_flags & (MDB_TXN_RDONLY|MDB_TXN_ERROR)) @@ -6077,6 +6371,9 @@ mdb_cursor_del(MDB_cursor *mc, unsigned int flags) if (!(mc->mc_flags & C_INITIALIZED)) return EINVAL; + if (mc->mc_ki[mc->mc_top] >= NUMKEYS(mc->mc_pg[mc->mc_top])) + return MDB_NOTFOUND; + if (!(flags & MDB_NOSPILL) && (rc = mdb_page_spill(mc, NULL, NULL))) return rc; @@ -6084,14 +6381,22 @@ mdb_cursor_del(MDB_cursor *mc, unsigned int flags) if (rc) return rc; - leaf = NODEPTR(mc->mc_pg[mc->mc_top], mc->mc_ki[mc->mc_top]); + 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(mc->mc_pg[mc->mc_top]) && 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) { @@ -6101,13 +6406,13 @@ mdb_cursor_del(MDB_cursor *mc, unsigned int flags) } else { MDB_cursor *m2; /* shrink fake page */ - mdb_node_shrink(mc->mc_pg[mc->mc_top], mc->mc_ki[mc->mc_top]); - leaf = NODEPTR(mc->mc_pg[mc->mc_top], mc->mc_ki[mc->mc_top]); + mdb_node_shrink(mp, mc->mc_ki[mc->mc_top]); + leaf = NODEPTR(mp, mc->mc_ki[mc->mc_top]); mc->mc_xcursor->mx_cursor.mc_pg[0] = NODEDATA(leaf); /* fix other sub-DB cursors pointed at this fake page */ for (m2 = mc->mc_txn->mt_cursors[mc->mc_dbi]; m2; m2=m2->mc_next) { if (m2 == mc || m2->mc_snum < mc->mc_snum) continue; - if (m2->mc_pg[mc->mc_top] == mc->mc_pg[mc->mc_top] && + if (m2->mc_pg[mc->mc_top] == mp && m2->mc_ki[mc->mc_top] == mc->mc_ki[mc->mc_top]) m2->mc_xcursor->mx_cursor.mc_pg[0] = NODEDATA(leaf); } @@ -6122,14 +6427,28 @@ mdb_cursor_del(MDB_cursor *mc, unsigned int flags) 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. @@ -6184,13 +6503,12 @@ mdb_leaf_size(MDB_env *env, MDB_val *key, MDB_val *data) 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. @@ -6209,7 +6527,7 @@ mdb_branch_size(MDB_env *env, MDB_val *key) 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); */ @@ -6246,12 +6564,12 @@ mdb_node_add(MDB_cursor *mc, indx_t indx, 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)) { @@ -6274,17 +6592,17 @@ mdb_node_add(MDB_cursor *mc, indx_t indx, 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))) @@ -6296,7 +6614,7 @@ mdb_node_add(MDB_cursor *mc, indx_t indx, node_size += data->mv_size; } } - node_size += node_size & 1; + node_size = EVEN(node_size); if ((ssize_t)node_size > room) goto full; @@ -6307,7 +6625,7 @@ update: /* 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); @@ -6325,7 +6643,7 @@ update: 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, @@ -6349,38 +6667,35 @@ update: 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); @@ -6397,10 +6712,9 @@ mdb_node_del(MDB_page *mp, indx_t indx, int 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]; @@ -6427,25 +6741,22 @@ mdb_node_shrink(MDB_page *mp, indx_t indx) 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; @@ -6456,6 +6767,9 @@ mdb_node_shrink(MDB_page *mp, indx_t indx) 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); @@ -6543,11 +6857,7 @@ mdb_xcursor_init1(MDB_cursor *mc, MDB_node *node) 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 } @@ -6567,7 +6877,7 @@ mdb_cursor_init(MDB_cursor *mc, MDB_txn *txn, MDB_dbi dbi, MDB_xcursor *mx) 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 { @@ -6584,7 +6894,7 @@ mdb_cursor_open(MDB_txn *txn, MDB_dbi dbi, MDB_cursor **ret) 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) @@ -6616,12 +6926,15 @@ mdb_cursor_open(MDB_txn *txn, MDB_dbi dbi, MDB_cursor **ret) 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; } @@ -6638,6 +6951,9 @@ mdb_cursor_count(MDB_cursor *mc, size_t *countp) 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; @@ -6675,11 +6991,10 @@ mdb_cursor_txn(MDB_cursor *mc) 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. @@ -6691,7 +7006,7 @@ mdb_update_key(MDB_cursor *mc, MDB_val *key) MDB_node *node; char *base; size_t len; - int delta, delta0; + int delta, ksize, oksize; indx_t ptr, i, numkeys, indx; DKBUF; @@ -6702,7 +7017,7 @@ mdb_update_key(MDB_cursor *mc, MDB_val *key) #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", @@ -6713,19 +7028,19 @@ mdb_update_key(MDB_cursor *mc, MDB_val *key) } #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); } @@ -6744,7 +7059,7 @@ mdb_update_key(MDB_cursor *mc, MDB_val *key) } /* 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) @@ -6776,7 +7091,6 @@ mdb_node_move(MDB_cursor *csrc, MDB_cursor *cdst) 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; @@ -6785,14 +7099,16 @@ mdb_node_move(MDB_cursor *csrc, MDB_cursor *cdst) 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); @@ -6815,18 +7131,20 @@ mdb_node_move(MDB_cursor *csrc, MDB_cursor *cdst) 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) @@ -6848,7 +7166,7 @@ mdb_node_move(MDB_cursor *csrc, MDB_cursor *cdst) /* 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 */ @@ -6896,7 +7214,7 @@ mdb_node_move(MDB_cursor *csrc, MDB_cursor *cdst) 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); } } @@ -6924,7 +7242,7 @@ mdb_node_move(MDB_cursor *csrc, MDB_cursor *cdst) 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); } } @@ -6937,21 +7255,25 @@ mdb_node_move(MDB_cursor *csrc, MDB_cursor *cdst) * 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; - DPRINTF(("merging page %"Z"u into %"Z"u", csrc->mc_pg[csrc->mc_top]->mp_pgno, - cdst->mc_pg[cdst->mc_top]->mp_pgno)); + psrc = csrc->mc_pg[csrc->mc_top]; + pdst = cdst->mc_pg[cdst->mc_top]; - assert(csrc->mc_snum > 1); /* can't merge root page */ - assert(cdst->mc_snum > 1); + DPRINTF(("merging page %"Z"u into %"Z"u", psrc->mp_pgno, pdst->mp_pgno)); + + 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))) @@ -6959,34 +7281,35 @@ mdb_page_merge(MDB_cursor *csrc, MDB_cursor *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); @@ -7001,26 +7324,35 @@ mdb_page_merge(MDB_cursor *csrc, MDB_cursor *cdst) } 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--; @@ -7028,7 +7360,6 @@ mdb_page_merge(MDB_cursor *csrc, MDB_cursor *cdst) /* 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) @@ -7037,15 +7368,24 @@ mdb_page_merge(MDB_cursor *csrc, MDB_cursor *cdst) 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. @@ -7083,27 +7423,18 @@ mdb_rebalance(MDB_cursor *mc) 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; } @@ -7143,6 +7474,7 @@ mdb_rebalance(MDB_cursor *mc) } } } 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) @@ -7154,6 +7486,10 @@ mdb_rebalance(MDB_cursor *mc) mc->mc_db->md_depth--; mc->mc_db->md_branch_pages--; mc->mc_ki[0] = mc->mc_ki[1]; + for (i = 1; imc_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; @@ -7166,7 +7502,6 @@ mdb_rebalance(MDB_cursor *mc) 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; imc_snum; i++) { @@ -7185,7 +7520,7 @@ mdb_rebalance(MDB_cursor *mc) * 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 @@ -7197,6 +7532,7 @@ mdb_rebalance(MDB_cursor *mc) 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. */ @@ -7230,113 +7566,115 @@ mdb_rebalance(MDB_cursor *mc) * (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) + 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_flags & C_INITIALIZED)) + 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; @@ -7344,6 +7682,7 @@ mdb_del(MDB_txn *txn, MDB_dbi dbi, } else { op = MDB_SET; xdata = NULL; + flags |= MDB_NODUPDATA; } rc = mdb_cursor_set(&mc, key, xdata, op, &exact); if (rc == 0) { @@ -7358,7 +7697,7 @@ mdb_del(MDB_txn *txn, MDB_dbi dbi, 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; @@ -7407,7 +7746,7 @@ mdb_page_split(MDB_cursor *mc, MDB_val *newkey, MDB_val *newdata, pgno_t newpgno 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]; @@ -7425,7 +7764,7 @@ mdb_page_split(MDB_cursor *mc, MDB_val *newkey, MDB_val *newdata, pgno_t newpgno 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; @@ -7454,7 +7793,6 @@ mdb_page_split(MDB_cursor *mc, MDB_val *newkey, MDB_val *newdata, pgno_t newpgno 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); @@ -7497,12 +7835,14 @@ mdb_page_split(MDB_cursor *mc, MDB_val *newkey, MDB_val *newdata, pgno_t newpgno 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; @@ -7536,7 +7876,7 @@ mdb_page_split(MDB_cursor *mc, MDB_val *newkey, MDB_val *newdata, pgno_t newpgno psize = 0; if (newindx <= split_indx || newindx >= nkeys) { i = 0; j = 1; - k = newindx >= nkeys ? nkeys : split_indx+1; + k = newindx >= nkeys ? nkeys : split_indx+2; } else { i = nkeys; j = -1; k = split_indx-1; @@ -7554,20 +7894,13 @@ mdb_page_split(MDB_cursor *mc, MDB_val *newkey, MDB_val *newdata, pgno_t newpgno else psize += NODEDSZ(node); } - psize += psize & 1; + psize = EVEN(psize); } - if (psize > pmax) { + if (psize > pmax || i == k-j) { split_indx = i + (j<0); break; } } - /* special case: when the new node was on the last - * slot we may not have tripped the break inside the loop. - * In all other cases we either hit the break condition, - * or the original split_indx was already safe. - */ - if (newindx >= nkeys && i == k) - split_indx = nkeys-1; } if (split_indx == newindx) { sepkey.mv_size = newkey->mv_size; @@ -7589,6 +7922,8 @@ mdb_page_split(MDB_cursor *mc, MDB_val *newkey, MDB_val *newdata, pgno_t newpgno 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) { @@ -7610,7 +7945,13 @@ mdb_page_split(MDB_cursor *mc, MDB_val *newkey, MDB_val *newdata, pgno_t newpgno 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--; @@ -7619,14 +7960,14 @@ mdb_page_split(MDB_cursor *mc, MDB_val *newkey, MDB_val *newdata, pgno_t newpgno } 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; imc_top; i++) mc->mc_ki[i] = mn.mc_ki[i]; } else if (!IS_LEAF2(mp)) { @@ -7664,11 +8005,8 @@ mdb_page_split(MDB_cursor *mc, MDB_val *newkey, MDB_val *newdata, pgno_t newpgno } 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; @@ -7702,16 +8040,12 @@ mdb_page_split(MDB_cursor *mc, MDB_val *newkey, MDB_val *newdata, pgno_t newpgno */ if (mn.mc_pg[ptop] != mc->mc_pg[ptop] && mc->mc_ki[ptop] >= NUMKEYS(mc->mc_pg[ptop])) { - for (i=0; imc_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); } { @@ -7762,6 +8096,12 @@ mdb_page_split(MDB_cursor *mc, MDB_val *newkey, MDB_val *newdata, pgno_t newpgno } } 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; } @@ -7772,10 +8112,7 @@ mdb_put(MDB_txn *txn, MDB_dbi dbi, 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) @@ -7807,6 +8144,32 @@ mdb_env_get_flags(MDB_env *env, unsigned int *arg) 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) { @@ -7817,6 +8180,16 @@ mdb_env_get_path(MDB_env *env, const char **arg) return MDB_SUCCESS; } +int +mdb_env_get_fd(MDB_env *env, mdb_filehandle_t *arg) +{ + if (!env || !arg) + return EINVAL; + + *arg = env->me_fd; + return MDB_SUCCESS; +} + /** Common code for #mdb_stat() and #mdb_env_stat(). * @param[in] env the environment to operate in. * @param[in] db the #MDB_db record containing the stats to return. @@ -7998,9 +8371,12 @@ int mdb_dbi_open(MDB_txn *txn, const char *name, unsigned int flags, MDB_dbi *db 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; @@ -8025,7 +8401,7 @@ void mdb_dbi_close(MDB_env *env, MDB_dbi dbi) 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; @@ -8065,22 +8441,22 @@ mdb_drop0(MDB_cursor *mc, int subs) 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; imc_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. */ @@ -8107,6 +8485,9 @@ mdb_drop0(MDB_cursor *mc, int subs) } /* 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; } @@ -8118,7 +8499,7 @@ int mdb_drop(MDB_txn *txn, MDB_dbi dbi, int del) 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)) @@ -8137,10 +8518,12 @@ int mdb_drop(MDB_txn *txn, MDB_dbi dbi, int del) /* 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 */ @@ -8161,7 +8544,7 @@ leave: 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; @@ -8170,7 +8553,7 @@ int mdb_set_compare(MDB_txn *txn, MDB_dbi dbi, MDB_cmp_func *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; @@ -8179,7 +8562,7 @@ int mdb_set_dupsort(MDB_txn *txn, MDB_dbi dbi, MDB_cmp_func *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; @@ -8188,7 +8571,7 @@ int mdb_set_relfunc(MDB_txn *txn, MDB_dbi dbi, MDB_rel_func *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; @@ -8197,7 +8580,7 @@ int mdb_set_relctx(MDB_txn *txn, MDB_dbi dbi, void *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) @@ -8205,7 +8588,7 @@ 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; @@ -8216,33 +8599,31 @@ int mdb_reader_list(MDB_env *env, MDB_msg_func *func, void *ctx) mr = env->me_txns->mti_readers; for (i=0; ime_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; ime_pid) { pid = mr[i].mr_pid;