X-Git-Url: https://git.sur5r.net/?a=blobdiff_plain;f=libraries%2Fliblmdb%2Fmdb.c;h=5acdba4dac75bd50f58d1123edcd3bf6df637cba;hb=3e98addbe294c69b94545141f466836bf92693ee;hp=e6ac29930dbed05183cac0921e2e429678eac681;hpb=1f191ba930938d6860e68f859fb1eb8374fbe799;p=openldap diff --git a/libraries/liblmdb/mdb.c b/libraries/liblmdb/mdb.c index e6ac29930d..5acdba4dac 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 @@ -35,15 +35,17 @@ #ifndef _GNU_SOURCE #define _GNU_SOURCE 1 #endif -#include -#include #ifdef _WIN32 +#include #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 +#define MDB_THR_T DWORD +#include +#include #ifdef __GNUC__ # include #else @@ -55,7 +57,10 @@ # endif #endif #else +#include +#include #define MDB_PID_T pid_t +#define MDB_THR_T pthread_t #include #include #include @@ -65,7 +70,6 @@ #include #endif -#include #include #include #include @@ -146,21 +150,34 @@ # 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 -#define pthread_t DWORD +#define THREAD_RET DWORD +#define pthread_t HANDLE #define pthread_mutex_t HANDLE +#define pthread_cond_t HANDLE #define pthread_key_t DWORD #define pthread_self() GetCurrentThreadId() #define pthread_key_create(x,y) \ @@ -170,6 +187,10 @@ #define pthread_setspecific(x,y) (TlsSetValue(x,y) ? 0 : ErrCode()) #define pthread_mutex_unlock(x) ReleaseMutex(x) #define pthread_mutex_lock(x) WaitForSingleObject(x, INFINITE) +#define pthread_cond_signal(x) SetEvent(*x) +#define pthread_cond_wait(cond,mutex) SignalObjectAndWait(*mutex, *cond, INFINITE, FALSE); WaitForSingleObject(*mutex, INFINITE) +#define THREAD_CREATE(thr,start,arg) thr=CreateThread(NULL,0,start,arg,0,NULL) +#define THREAD_FINISH(thr) WaitForSingleObject(thr, INFINITE) #define LOCK_MUTEX_R(env) pthread_mutex_lock((env)->me_rmutex) #define UNLOCK_MUTEX_R(env) pthread_mutex_unlock((env)->me_rmutex) #define LOCK_MUTEX_W(env) pthread_mutex_lock((env)->me_wmutex) @@ -188,7 +209,9 @@ #endif #define Z "I" #else - +#define THREAD_RET void * +#define THREAD_CREATE(thr,start,arg) pthread_create(&thr,NULL,start,arg) +#define THREAD_FINISH(thr) pthread_join(thr,NULL) #define Z "z" /**< printf format modifier for size_t */ /** For MDB_LOCK_FORMAT: True if readers take a pid lock in the lockfile */ @@ -327,7 +350,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 @@ -342,7 +365,7 @@ static txnid_t mdb_debug_start; /** @brief The maximum size of a database page. * - * This is 32k, since it must fit in #MDB_page.#mp_upper. + * 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 @@ -371,7 +394,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. */ @@ -527,7 +550,7 @@ typedef struct MDB_rxbody { /** The process ID of the process owning this reader txn. */ MDB_PID_T mrb_pid; /** The thread ID of the thread owning this txn. */ - pthread_t mrb_tid; + MDB_THR_T mrb_tid; } MDB_rxbody; /** The actual reader record, with cacheline padding. */ @@ -558,7 +581,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. */ @@ -640,6 +663,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 */ @@ -690,6 +714,9 @@ typedef struct MDB_page { /** The number of overflow pages needed to store the given size. */ #define OVPAGES(size, psize) ((PAGEHDRSZ-1 + (size)) / (psize) + 1) + /** Link in #MDB_txn.%mt_loose_pages list */ +#define NEXT_LOOSE_PAGE(p) (*(MDB_page **)METADATA(p)) + /** Header for a single key/data pair within a page. * Used in pages of type #P_BRANCH and #P_LEAF without #P_LEAF2. * We guarantee 2-byte alignment for 'MDB_node's. @@ -831,7 +858,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. */ @@ -888,6 +915,10 @@ struct MDB_txn { /** The list of pages that became unused during this transaction. */ MDB_IDL mt_free_pgs; + /** The list of loose pages that became unused and may be reused + * in this transaction, linked through #NEXT_LOOSE_PAGE(page). + */ + MDB_page *mt_loose_pgs; /** The sorted list of dirty pages we temporarily wrote to disk * because the dirty list was full. page numbers in here are * shifted left by 1, deleted slots have the LSB set. @@ -926,12 +957,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. @@ -1024,8 +1055,6 @@ 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; /**< DB page size, inited from me_os_psize */ unsigned int me_os_psize; /**< OS page size, from #GET_PAGESIZE */ @@ -1061,6 +1090,7 @@ struct MDB_env { #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 */ @@ -1069,6 +1099,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 */ @@ -1084,9 +1116,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); @@ -1117,7 +1153,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); @@ -1130,7 +1166,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); @@ -1166,7 +1203,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", @@ -1174,7 +1211,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", @@ -1186,7 +1223,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 * @@ -1204,6 +1241,33 @@ 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 @@ -1244,18 +1308,55 @@ 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, 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", mdb_dbg_pgno(mp), 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; @@ -1271,11 +1372,13 @@ mdb_page_list(MDB_page *mp) nsize += NODEDSZ(node); total += nsize; nsize += sizeof(indx_t); - fprintf(stderr, "key %d: nsize %d, %s\n", i, nsize, DKEY(&key)); + fprintf(stderr, "key %d: nsize %d, %s%s\n", + i, nsize, DKEY(&key), mdb_leafnode_type(node)); } total = EVEN(total); } - fprintf(stderr, "Total: %d\n", total); + fprintf(stderr, "Total: header %d + contents %d + unused %d\n", + IS_LEAF2(mp) ? PAGEHDRSZ : mp->mp_lower, total, SIZELEFT(mp)); } void @@ -1301,6 +1404,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) { @@ -1314,10 +1418,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; @@ -1325,8 +1432,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]; @@ -1340,7 +1447,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) { @@ -1394,10 +1501,11 @@ mdb_page_malloc(MDB_txn *txn, unsigned num) 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.) @@ -1437,6 +1545,58 @@ mdb_dlist_free(MDB_txn *txn) dl[0].mid = 0; } +/** Loosen or free 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. + * + * If the page wasn't dirtied in this txn, just add it + * to this txn's free list. + */ +static int +mdb_page_loose(MDB_cursor *mc, MDB_page *mp) +{ + int loose = 0; + pgno_t pgno = mp->mp_pgno; + + if ((mp->mp_flags & P_DIRTY) && mc->mc_dbi != FREE_DBI) { + if (mc->mc_txn->mt_parent) { + MDB_ID2 *dl = mc->mc_txn->mt_u.dirty_list; + /* If txn has a parent, make sure the page is in our + * dirty list. + */ + if (dl[0].mid) { + unsigned x = mdb_mid2l_search(dl, pgno); + if (x <= dl[0].mid && dl[x].mid == pgno) { + if (mp != dl[x].mptr) { /* bad cursor? */ + mc->mc_flags &= ~(C_INITIALIZED|C_EOF); + mc->mc_txn->mt_flags |= MDB_TXN_ERROR; + return MDB_CORRUPTED; + } + /* ok, it's ours */ + loose = 1; + } + } + } else { + /* no parent txn, so it's just ours */ + loose = 1; + } + } + if (loose) { + NEXT_LOOSE_PAGE(mp) = mc->mc_txn->mt_loose_pgs; + mc->mc_txn->mt_loose_pgs = mp; + mp->mp_flags |= P_LOOSE; + } else { + int rc = mdb_midl_append(&mc->mc_txn->mt_free_pgs, pgno); + if (rc) + return rc; + } + + return MDB_SUCCESS; +} + /** 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: @@ -1485,6 +1645,12 @@ mdb_pages_xkeep(MDB_cursor *mc, unsigned pflags, int all) break; } + /* Loose pages shouldn't be spilled */ + for (dp = txn->mt_loose_pgs; dp; dp = NEXT_LOOSE_PAGE(dp)) { + if ((dp->mp_flags & Mask) == pflags) + dp->mp_flags ^= P_KEEP; + } + if (all) { /* Mark dirty root pages */ for (i=0; imt_numdbs; i++) { @@ -1659,7 +1825,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; @@ -1668,7 +1834,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--; } @@ -1701,7 +1868,7 @@ mdb_page_alloc(MDB_cursor *mc, int num, MDB_page **mp) #else enum { Paranoid = 0, Max_retries = INT_MAX /*infinite*/ }; #endif - int rc, 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; @@ -1711,11 +1878,21 @@ mdb_page_alloc(MDB_cursor *mc, int num, MDB_page **mp) MDB_cursor_op op; MDB_cursor m2; + /* If there are any loose pages, just use them */ + if (num == 1 && txn->mt_loose_pgs) { + np = txn->mt_loose_pgs; + txn->mt_loose_pgs = NEXT_LOOSE_PAGE(np); + *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; @@ -1732,7 +1909,7 @@ mdb_page_alloc(MDB_cursor *mc, int num, MDB_page **mp) if (mop[i-n2] == pgno+n2) goto search_done; } while (--i > n2); - if (Max_retries < INT_MAX && --retry < 0) + if (--retry < 0) break; } @@ -1760,7 +1937,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) @@ -1773,11 +1950,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; @@ -1806,15 +1985,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; @@ -1829,6 +2011,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. @@ -1935,17 +2121,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) { @@ -1966,19 +2152,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; } @@ -2004,6 +2192,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]); @@ -2014,6 +2203,10 @@ done: } } return 0; + +fail: + txn->mt_flags |= MDB_TXN_ERROR; + return rc; } int @@ -2207,13 +2400,13 @@ mdb_txn_renew0(MDB_txn *txn) return MDB_BAD_RSLOT; } else { MDB_PID_T pid = env->me_pid; - pthread_t tid = pthread_self(); + MDB_THR_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) return rc; - env->me_flags |= MDB_LIVE_READER; + env->me_live_reader = 1; } LOCK_MUTEX_R(env); @@ -2554,6 +2747,32 @@ 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 (txn->mt_loose_pgs) { + MDB_page *mp = txn->mt_loose_pgs; + /* Just return them to freeDB */ + if (env->me_pghead) { + int i, j; + mop = env->me_pghead; + for (; mp; mp = NEXT_LOOSE_PAGE(mp)) { + pgno_t pg = mp->mp_pgno; + j = mop[0] + 1; + for (i = mop[0]; i && mop[i] < pg; i--) + mop[j--] = mop[i]; + mop[j] = pg; + mop[0] += 1; + } + } else { + /* Oh well, they were wasted. Put on freelist */ + for (; mp; mp = NEXT_LOOSE_PAGE(mp)) { + mdb_midl_append(&txn->mt_free_pgs, mp->mp_pgno); + } + } + txn->mt_loose_pgs = NULL; + } + /* MDB_RESERVE cancels meminit in ovpage malloc (when no WRITEMAP) */ clean_limit = (env->me_flags & (MDB_NOMEMINIT|MDB_WRITEMAP)) ? SSIZE_MAX : maxfree_1pg; @@ -2573,7 +2792,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; @@ -2663,22 +2882,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; @@ -2834,8 +3051,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); @@ -2943,7 +3160,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; @@ -3097,6 +3314,20 @@ mdb_env_read_header(MDB_env *env, MDB_meta *meta) return 0; } +static void +mdb_env_init_meta0(MDB_env *env, MDB_meta *meta) +{ + meta->mm_magic = MDB_MAGIC; + meta->mm_version = MDB_DATA_VERSION; + meta->mm_mapsize = env->me_mapsize; + meta->mm_psize = env->me_psize; + meta->mm_last_pg = 1; + meta->mm_flags = env->me_flags & 0xffff; + meta->mm_flags |= MDB_INTEGERKEY; + meta->mm_dbs[0].md_root = P_INVALID; + meta->mm_dbs[1].md_root = P_INVALID; +} + /** Write the environment parameters of a freshly created DB environment. * @param[in] env the environment handle * @param[out] meta address of where to store the meta information @@ -3126,15 +3357,7 @@ mdb_env_init_meta(MDB_env *env, MDB_meta *meta) psize = env->me_psize; - meta->mm_magic = MDB_MAGIC; - meta->mm_version = MDB_DATA_VERSION; - meta->mm_mapsize = env->me_mapsize; - meta->mm_psize = psize; - meta->mm_last_pg = 1; - meta->mm_flags = env->me_flags & 0xffff; - meta->mm_flags |= MDB_INTEGERKEY; - meta->mm_dbs[0].md_root = P_INVALID; - meta->mm_dbs[1].md_root = P_INVALID; + mdb_env_init_meta0(env, meta); p = calloc(2, psize); p->mp_pgno = 0; @@ -3176,9 +3399,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)); @@ -3327,8 +3547,17 @@ mdb_env_map(MDB_env *env, void *addr, int newsize) int rc; HANDLE mh; LONG sizelo, sizehi; - sizelo = env->me_mapsize & 0xffffffff; - sizehi = env->me_mapsize >> 16 >> 16; /* only needed on Win64 */ + size_t msize; + + if (flags & MDB_RDONLY) { + msize = 0; + sizelo = 0; + sizehi = 0; + } else { + msize = env->me_mapsize; + sizelo = msize & 0xffffffff; + sizehi = msize >> 16 >> 16; /* only needed on Win64 */ + } /* Windows won't create mappings for zero length files. * Just allocate the maxsize right now. @@ -3346,7 +3575,7 @@ mdb_env_map(MDB_env *env, void *addr, int newsize) return ErrCode(); env->me_map = MapViewOfFileEx(mh, flags & MDB_WRITEMAP ? FILE_MAP_WRITE : FILE_MAP_READ, - 0, 0, env->me_mapsize, addr); + 0, 0, msize, addr); rc = env->me_map ? 0 : ErrCode(); CloseHandle(mh); if (rc) @@ -3453,7 +3682,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) @@ -3500,7 +3729,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; @@ -3575,7 +3804,9 @@ static void NTAPI mdb_tls_callback(PVOID module, DWORD reason, PVOID ptr) case DLL_THREAD_DETACH: for (i=0; ime_flags &= ~(MDB_ENV_ACTIVE|MDB_ENV_TXKEY); } -int -mdb_env_copyfd(MDB_env *env, HANDLE fd) -{ - MDB_txn *txn = NULL; - int rc; - size_t wsize; - char *ptr; -#ifdef _WIN32 - DWORD len, w2; -#define DO_WRITE(rc, fd, ptr, w2, len) rc = WriteFile(fd, ptr, w2, &len, NULL) -#else - ssize_t len; - size_t w2; -#define DO_WRITE(rc, fd, ptr, w2, len) len = write(fd, ptr, w2); rc = (len >= 0) -#endif - - /* Do the lock/unlock of the reader mutex before starting the - * write txn. Otherwise other read txns could block writers. - */ - rc = mdb_txn_begin(env, NULL, MDB_RDONLY, &txn); - if (rc) - return rc; - - if (env->me_txns) { - /* We must start the actual read txn after blocking writers */ - mdb_txn_reset0(txn, "reset-stage1"); - /* Temporarily block writers until we snapshot the meta pages */ - LOCK_MUTEX_W(env); +void +mdb_env_close(MDB_env *env) +{ + MDB_page *dp; - rc = mdb_txn_renew0(txn); - if (rc) { - UNLOCK_MUTEX_W(env); - goto leave; - } - } + if (env == NULL) + return; - wsize = env->me_psize * 2; - ptr = env->me_map; - w2 = wsize; - while (w2 > 0) { - DO_WRITE(rc, fd, ptr, w2, len); - if (!rc) { - rc = ErrCode(); - break; - } else if (len > 0) { - rc = MDB_SUCCESS; - ptr += len; - w2 -= len; - continue; - } else { - /* Non-blocking or async handles are not supported */ - rc = EIO; - break; - } + VGMEMP_DESTROY(env); + while ((dp = env->me_dpages) != NULL) { + VGMEMP_DEFINED(&dp->mp_next, sizeof(dp->mp_next)); + env->me_dpages = dp->mp_next; + free(dp); } - if (env->me_txns) - UNLOCK_MUTEX_W(env); - if (rc) - goto leave; + mdb_env_close0(env, 0); + free(env); +} - wsize = txn->mt_next_pgno * env->me_psize - wsize; - while (wsize > 0) { - if (wsize > MAX_WRITE) - w2 = MAX_WRITE; - else - w2 = wsize; - DO_WRITE(rc, fd, ptr, w2, len); - if (!rc) { - rc = ErrCode(); - break; - } else if (len > 0) { - rc = MDB_SUCCESS; - ptr += len; - wsize -= len; - continue; - } else { - rc = EIO; - break; - } - } +/** Compare two items pointing at aligned size_t's */ +static int +mdb_cmp_long(const MDB_val *a, const MDB_val *b) +{ + return (*(size_t *)a->mv_data < *(size_t *)b->mv_data) ? -1 : + *(size_t *)a->mv_data > *(size_t *)b->mv_data; +} -leave: - mdb_txn_abort(txn); - return rc; +/** Compare two items pointing at aligned unsigned int's */ +static int +mdb_cmp_int(const MDB_val *a, const MDB_val *b) +{ + return (*(unsigned int *)a->mv_data < *(unsigned int *)b->mv_data) ? -1 : + *(unsigned int *)a->mv_data > *(unsigned int *)b->mv_data; } -int -mdb_env_copy(MDB_env *env, const char *path) +/** Compare two items pointing at unsigned ints of unknown alignment. + * Nodes and keys are guaranteed to be 2-byte aligned. + */ +static int +mdb_cmp_cint(const MDB_val *a, const MDB_val *b) { - int rc, len; - char *lpath; - HANDLE newfd = INVALID_HANDLE_VALUE; - - if (env->me_flags & MDB_NOSUBDIR) { - lpath = (char *)path; - } else { - len = strlen(path); - len += sizeof(DATANAME); - lpath = malloc(len); - if (!lpath) - return ENOMEM; - sprintf(lpath, "%s" DATANAME, path); - } - - /* The destination path must exist, but the destination file must not. - * We don't want the OS to cache the writes, since the source data is - * already in the OS cache. - */ -#ifdef _WIN32 - newfd = CreateFile(lpath, GENERIC_WRITE, 0, NULL, CREATE_NEW, - FILE_FLAG_NO_BUFFERING|FILE_FLAG_WRITE_THROUGH, NULL); -#else - newfd = open(lpath, O_WRONLY|O_CREAT|O_EXCL, 0666); -#endif - if (newfd == INVALID_HANDLE_VALUE) { - rc = ErrCode(); - goto leave; - } - -#ifdef O_DIRECT - /* Set O_DIRECT if the file system supports it */ - if ((rc = fcntl(newfd, F_GETFL)) != -1) - (void) fcntl(newfd, F_SETFL, rc | O_DIRECT); -#endif -#ifdef F_NOCACHE /* __APPLE__ */ - rc = fcntl(newfd, F_NOCACHE, 1); - if (rc) { - rc = ErrCode(); - goto leave; - } -#endif - - rc = mdb_env_copyfd(env, newfd); - -leave: - if (!(env->me_flags & MDB_NOSUBDIR)) - free(lpath); - if (newfd != INVALID_HANDLE_VALUE) - if (close(newfd) < 0 && rc == MDB_SUCCESS) - rc = ErrCode(); - - return rc; -} - -void -mdb_env_close(MDB_env *env) -{ - MDB_page *dp; - - if (env == NULL) - return; - - VGMEMP_DESTROY(env); - while ((dp = env->me_dpages) != NULL) { - VGMEMP_DEFINED(&dp->mp_next, sizeof(dp->mp_next)); - env->me_dpages = dp->mp_next; - free(dp); - } - - mdb_env_close0(env, 0); - free(env); -} - -/** Compare two items pointing at aligned size_t's */ -static int -mdb_cmp_long(const MDB_val *a, const MDB_val *b) -{ - return (*(size_t *)a->mv_data < *(size_t *)b->mv_data) ? -1 : - *(size_t *)a->mv_data > *(size_t *)b->mv_data; -} - -/** Compare two items pointing at aligned unsigned int's */ -static int -mdb_cmp_int(const MDB_val *a, const MDB_val *b) -{ - return (*(unsigned int *)a->mv_data < *(unsigned int *)b->mv_data) ? -1 : - *(unsigned int *)a->mv_data > *(unsigned int *)b->mv_data; -} - -/** Compare two items pointing at unsigned ints of unknown alignment. - * Nodes and keys are guaranteed to be 2-byte aligned. - */ -static int -mdb_cmp_cint(const MDB_val *a, const MDB_val *b) -{ -#if BYTE_ORDER == LITTLE_ENDIAN - unsigned short *u, *c; - int x; +#if BYTE_ORDER == LITTLE_ENDIAN + unsigned short *u, *c; + int x; u = (unsigned short *) ((char *) a->mv_data + a->mv_size); c = (unsigned short *) ((char *) b->mv_data + a->mv_size); @@ -4438,10 +4524,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) @@ -4624,7 +4726,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; } @@ -4684,7 +4786,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; } @@ -4710,7 +4812,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)) { @@ -4725,14 +4827,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) @@ -4752,6 +4854,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; } @@ -4846,7 +4949,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; @@ -4916,7 +5019,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; @@ -4984,11 +5087,9 @@ 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) @@ -5039,7 +5140,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) { @@ -5067,7 +5168,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]; @@ -5115,7 +5216,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)) { @@ -5144,7 +5245,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]; @@ -5191,7 +5292,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)) { @@ -5222,8 +5323,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; @@ -5313,7 +5412,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 @@ -5326,7 +5425,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); @@ -5340,7 +5439,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); } @@ -5349,8 +5448,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; } @@ -5416,7 +5517,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; @@ -5462,7 +5563,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; @@ -5499,7 +5600,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; @@ -5691,19 +5793,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 = mc->mc_txn->mt_env; + MDB_env *env; MDB_node *leaf = NULL; MDB_page *fp, *mp; uint16_t fp_flags; MDB_val xdata, *rdata, dkey, olddata; MDB_db dummy; - int do_sub = 0, insert; + int do_sub = 0, insert_key, insert_data; unsigned int mcount = 0, dcount = 0, nospill; size_t nsize; int rc, rc2; 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. */ @@ -5720,7 +5827,7 @@ 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-1 >= ENV_MAXKEY(env)) + if (key->mv_size-1 >= ENV_MAXKEY(env)) return MDB_BAD_VALSIZE; #if SIZE_MAX > MAXDATASIZE @@ -5811,8 +5918,8 @@ mdb_cursor_put(MDB_cursor *mc, MDB_val *key, MDB_val *data, return rc2; } - insert = rc; - if (insert) { + 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) && @@ -5828,13 +5935,12 @@ mdb_cursor_put(MDB_cursor *mc, MDB_val *key, MDB_val *data, } 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; } @@ -5850,7 +5956,7 @@ more: * it. mp: new (sub-)page. offset: growth in page * size. xdata: node data with new page or DB. */ - ssize_t i, offset = 0; + unsigned i, offset = 0; mp = fp = xdata.mv_data = env->me_pbuf; mp->mp_pgno = mc->mc_pg[mc->mc_top]->mp_pgno; @@ -5862,21 +5968,14 @@ more: #if UINT_MAX < SIZE_MAX if (mc->mc_dbx->md_dcmp == mdb_cmp_int && olddata.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; + mc->mc_dbx->md_dcmp = mdb_cmp_clong; #endif -#endif - /* if data matches, skip it */ + /* 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; } /* Back up original data item */ @@ -5906,17 +6005,17 @@ more: fp = olddata.mv_data; switch (flags) { default: - i = -(ssize_t)SIZELEFT(fp); if (!(mc->mc_db->md_flags & MDB_DUPFIXED)) { - offset = i += (ssize_t) EVEN( - sizeof(indx_t) + NODESIZE + data->mv_size); - } else { - i += offset = fp->mp_pad; - offset *= 4; /* space for 4 more */ + offset = EVEN(NODESIZE + sizeof(indx_t) + + data->mv_size); + break; } - if (i > 0) + offset = fp->mp_pad; + if (SIZELEFT(fp) < offset) { + offset *= 4; /* space for 4 more */ break; - /* FALLTHRU: Sub-page is big enough */ + } + /* FALLTHRU: Big enough MDB_DUPFIXED sub-page */ case MDB_CURRENT: fp->mp_flags |= P_DIRTY; COPY_PGNO(fp->mp_pgno, mp->mp_pgno); @@ -5965,7 +6064,7 @@ prep_subDB: } else { memcpy((char *)mp + mp->mp_upper, (char *)fp + fp->mp_upper, olddata.mv_size - fp->mp_upper); - for (i = NUMKEYS(fp); --i >= 0; ) + for (i=0; imp_ptrs[i] = fp->mp_ptrs[i] + offset; } } @@ -5973,8 +6072,8 @@ prep_subDB: rdata = &xdata; flags |= F_DUPDATA; do_sub = 1; - if (!insert) - mdb_node_del(mc->mc_pg[mc->mc_top], mc->mc_ki[mc->mc_top], 0); + if (!insert_key) + mdb_node_del(mc, 0); goto new_sub; } current: @@ -6014,7 +6113,8 @@ current: 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. @@ -6032,7 +6132,7 @@ 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) @@ -6044,14 +6144,13 @@ current: */ if (F_ISSET(flags, MDB_RESERVE)) data->mv_data = olddata.mv_data; - else if (data->mv_size) + 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--; + mdb_node_del(mc, 0); } rdata = data; @@ -6061,14 +6160,14 @@ new_sub: 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; @@ -6088,9 +6187,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 @@ -6098,6 +6195,7 @@ new_sub: */ if (do_sub) { int xflags; + size_t ecount; put_sub: xdata.mv_size = 0; xdata.mv_data = ""; @@ -6113,7 +6211,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; @@ -6131,6 +6229,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); @@ -6138,31 +6237,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; + 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; } @@ -6190,14 +6296,21 @@ mdb_cursor_del(MDB_cursor *mc, unsigned int flags) return rc; mp = mc->mc_pg[mc->mc_top]; + if (IS_LEAF2(mp)) + goto del_key; leaf = NODEPTR(mp, mc->mc_ki[mc->mc_top]); - if (!IS_LEAF2(mp) && F_ISSET(leaf->mn_flags, F_DUPDATA)) { - if (!(flags & MDB_NODUPDATA)) { + if (F_ISSET(leaf->mn_flags, F_DUPDATA)) { + if (flags & MDB_NODUPDATA) { + /* mdb_cursor_del0() will subtract the final entry */ + mc->mc_db->md_entries -= mc->mc_xcursor->mx_db.md_entries - 1; + } else { if (!F_ISSET(leaf->mn_flags, F_SUBDATA)) { mc->mc_xcursor->mx_cursor.mc_pg[0] = NODEDATA(leaf); } rc = mdb_cursor_del(&mc->mc_xcursor->mx_cursor, MDB_NOSPILL); + if (rc) + return rc; /* If sub-DB still has entries, we're done */ if (mc->mc_xcursor->mx_db.md_entries) { if (leaf->mn_flags & F_SUBDATA) { @@ -6228,14 +6341,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. @@ -6351,7 +6478,7 @@ 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", @@ -6379,7 +6506,7 @@ 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); @@ -6412,7 +6539,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); @@ -6430,7 +6557,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, @@ -6457,18 +6584,20 @@ full: 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; @@ -6476,10 +6605,11 @@ mdb_node_del(MDB_page *mp, indx_t indx, int ksize) DPRINTF(("delete node %u on %s page %"Z"u", indx, IS_LEAF(mp) ? "leaf" : "branch", mdb_dbg_pgno(mp))); - assert(indx < NUMKEYS(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); @@ -6499,7 +6629,6 @@ mdb_node_del(MDB_page *mp, indx_t indx, int ksize) 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]; @@ -6642,11 +6771,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 } @@ -6666,7 +6791,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 { @@ -6683,7 +6808,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) @@ -6715,12 +6840,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; } @@ -6737,6 +6865,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; @@ -6774,7 +6905,6 @@ mdb_cursor_txn(MDB_cursor *mc) MDB_dbi mdb_cursor_dbi(MDB_cursor *mc) { - assert(mc != NULL); return mc->mc_dbi; } @@ -6824,7 +6954,7 @@ mdb_update_key(MDB_cursor *mc, MDB_val *key) /* 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); } @@ -6883,14 +7013,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(!((size_t)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); @@ -6913,18 +7045,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) @@ -6946,7 +7080,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 */ @@ -6994,7 +7128,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); } } @@ -7022,7 +7156,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); } } @@ -7035,21 +7169,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; + + psrc = csrc->mc_pg[csrc->mc_top]; + pdst = cdst->mc_pg[cdst->mc_top]; - DPRINTF(("merging page %"Z"u into %"Z"u", csrc->mc_pg[csrc->mc_top]->mp_pgno, - cdst->mc_pg[cdst->mc_top]->mp_pgno)); + DPRINTF(("merging page %"Z"u into %"Z"u", psrc->mp_pgno, pdst->mp_pgno)); - assert(csrc->mc_snum > 1); /* can't merge root page */ - assert(cdst->mc_snum > 1); + mdb_cassert(csrc, csrc->mc_snum > 1); /* can't merge root page */ + mdb_cassert(csrc, cdst->mc_snum > 1); /* Mark dst as dirty. */ if ((rc = mdb_page_touch(cdst))) @@ -7057,34 +7195,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); @@ -7099,26 +7238,31 @@ 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); + psrc = csrc->mc_pg[csrc->mc_top]; + /* If not operating on FreeDB, allow this page to be reused + * in this txn. Otherwise just add to free list. + */ + rc = mdb_page_loose(csrc, psrc); if (rc) return rc; - if (IS_LEAF(csrc->mc_pg[csrc->mc_top])) + if (IS_LEAF(psrc)) csrc->mc_db->md_leaf_pages--; else csrc->mc_db->md_branch_pages--; @@ -7126,7 +7270,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) @@ -7135,15 +7278,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. @@ -7181,6 +7333,7 @@ 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])); DPRINTF(("rebalancing %s page %"Z"u (has %u keys, %.1f%% full)", @@ -7231,6 +7384,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) @@ -7242,6 +7396,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; @@ -7254,7 +7412,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++) { @@ -7273,7 +7430,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 @@ -7285,6 +7442,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. */ @@ -7318,49 +7476,41 @@ 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 { + + if (rc == MDB_SUCCESS) { MDB_cursor *m2, *m3; MDB_dbi dbi = mc->mc_dbi; @@ -7368,11 +7518,14 @@ mdb_cursor_del0(MDB_cursor *mc, MDB_node *leaf) 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) { + 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; @@ -7384,44 +7537,54 @@ mdb_cursor_del0(MDB_cursor *mc, MDB_node *leaf) if (m3->mc_ki[mc->mc_top] > ki) m3->mc_ki[mc->mc_top]--; } - if (m3->mc_ki[mc->mc_top] >= nkeys) - mdb_cursor_sibling(m3, 1); + if (m3->mc_ki[mc->mc_top] >= nkeys) { + rc = mdb_cursor_sibling(m3, 1); + if (rc == MDB_NOTFOUND) + rc = MDB_SUCCESS; + } } } 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; - 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; @@ -7429,6 +7592,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) { @@ -7443,7 +7607,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; @@ -7492,7 +7656,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]; @@ -7510,7 +7674,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; @@ -7539,7 +7703,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); @@ -7586,8 +7749,10 @@ mdb_page_split(MDB_cursor *mc, MDB_val *newkey, MDB_val *newdata, pgno_t newpgno /* 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; @@ -7667,6 +7832,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) { @@ -7688,7 +7855,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--; @@ -7697,14 +7870,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)) { @@ -7742,11 +7915,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; @@ -7780,16 +7950,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); } { @@ -7840,6 +8006,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; } @@ -7850,10 +8022,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) @@ -7863,6 +8032,575 @@ mdb_put(MDB_txn *txn, MDB_dbi dbi, return mdb_cursor_put(&mc, key, data, flags); } +#ifndef MDB_WBUF +#define MDB_WBUF (1024*1024) +#endif + + /** State needed for a compacting copy. */ +typedef struct mdb_copy { + pthread_mutex_t mc_mutex; + pthread_cond_t mc_cond; + char *mc_wbuf[2]; + char *mc_over[2]; + MDB_env *mc_env; + MDB_txn *mc_txn; + int mc_wlen[2]; + int mc_olen[2]; + pgno_t mc_next_pgno; + HANDLE mc_fd; + int mc_status; + volatile int mc_new; + int mc_toggle; + +} mdb_copy; + + /** Dedicated writer thread for compacting copy. */ +static THREAD_RET +mdb_env_copythr(void *arg) +{ + mdb_copy *my = arg; + char *ptr; + int toggle = 0, wsize, rc; +#ifdef _WIN32 + DWORD len; +#define DO_WRITE(rc, fd, ptr, w2, len) rc = WriteFile(fd, ptr, w2, &len, NULL) +#else + int len; +#define DO_WRITE(rc, fd, ptr, w2, len) len = write(fd, ptr, w2); rc = (len >= 0) +#endif + + pthread_mutex_lock(&my->mc_mutex); + my->mc_new = 0; + pthread_cond_signal(&my->mc_cond); + for(;;) { + while (!my->mc_new) + pthread_cond_wait(&my->mc_cond, &my->mc_mutex); + if (my->mc_new < 0) { + my->mc_new = 0; + break; + } + my->mc_new = 0; + wsize = my->mc_wlen[toggle]; + ptr = my->mc_wbuf[toggle]; +again: + while (wsize > 0) { + DO_WRITE(rc, my->mc_fd, ptr, wsize, len); + if (!rc) { + rc = ErrCode(); + break; + } else if (len > 0) { + rc = MDB_SUCCESS; + ptr += len; + wsize -= len; + continue; + } else { + rc = EIO; + break; + } + } + if (rc) { + my->mc_status = rc; + break; + } + /* If there's an overflow page tail, write it too */ + if (my->mc_olen[toggle]) { + wsize = my->mc_olen[toggle]; + ptr = my->mc_over[toggle]; + my->mc_olen[toggle] = 0; + goto again; + } + my->mc_wlen[toggle] = 0; + toggle ^= 1; + pthread_cond_signal(&my->mc_cond); + } + pthread_cond_signal(&my->mc_cond); + pthread_mutex_unlock(&my->mc_mutex); + return (THREAD_RET)0; +#undef DO_WRITE +} + + /** Tell the writer thread there's a buffer ready to write */ +static int +mdb_env_cthr_toggle(mdb_copy *my, int st) +{ + int toggle = my->mc_toggle ^ 1; + pthread_mutex_lock(&my->mc_mutex); + if (my->mc_status) { + pthread_mutex_unlock(&my->mc_mutex); + return my->mc_status; + } + while (my->mc_new == 1) + pthread_cond_wait(&my->mc_cond, &my->mc_mutex); + my->mc_new = st; + my->mc_toggle = toggle; + pthread_cond_signal(&my->mc_cond); + pthread_mutex_unlock(&my->mc_mutex); + return 0; +} + + /** Depth-first tree traversal for compacting copy. */ +static int +mdb_env_cwalk(mdb_copy *my, pgno_t *pg, int flags) +{ + MDB_cursor mc; + MDB_txn *txn = my->mc_txn; + MDB_node *ni; + MDB_page *mo, *mp, *leaf; + char *buf, *ptr; + int rc, toggle; + unsigned int i; + + /* Empty DB, nothing to do */ + if (*pg == P_INVALID) + return MDB_SUCCESS; + + mc.mc_snum = 1; + mc.mc_top = 0; + mc.mc_txn = txn; + + rc = mdb_page_get(my->mc_txn, *pg, &mc.mc_pg[0], NULL); + if (rc) + return rc; + rc = mdb_page_search_root(&mc, NULL, MDB_PS_FIRST); + if (rc) + return rc; + + /* Make cursor pages writable */ + buf = ptr = malloc(my->mc_env->me_psize * mc.mc_snum); + if (buf == NULL) + return ENOMEM; + + for (i=0; imc_env->me_psize); + mc.mc_pg[i] = (MDB_page *)ptr; + ptr += my->mc_env->me_psize; + } + + /* This is writable space for a leaf page. Usually not needed. */ + leaf = (MDB_page *)ptr; + + toggle = my->mc_toggle; + while (mc.mc_snum > 0) { + unsigned n; + mp = mc.mc_pg[mc.mc_top]; + n = NUMKEYS(mp); + + if (IS_LEAF(mp)) { + if (!IS_LEAF2(mp) && !(flags & F_DUPDATA)) { + for (i=0; imn_flags & F_BIGDATA) { + MDB_page *omp; + pgno_t pg; + + /* Need writable leaf */ + if (mp != leaf) { + mc.mc_pg[mc.mc_top] = leaf; + mdb_page_copy(leaf, mp, my->mc_env->me_psize); + mp = leaf; + ni = NODEPTR(mp, i); + } + + memcpy(&pg, NODEDATA(ni), sizeof(pg)); + rc = mdb_page_get(txn, pg, &omp, NULL); + if (rc) + goto done; + if (my->mc_wlen[toggle] >= MDB_WBUF) { + rc = mdb_env_cthr_toggle(my, 1); + if (rc) + goto done; + toggle = my->mc_toggle; + } + mo = (MDB_page *)(my->mc_wbuf[toggle] + my->mc_wlen[toggle]); + memcpy(mo, omp, my->mc_env->me_psize); + mo->mp_pgno = my->mc_next_pgno; + my->mc_next_pgno += omp->mp_pages; + my->mc_wlen[toggle] += my->mc_env->me_psize; + if (omp->mp_pages > 1) { + my->mc_olen[toggle] = my->mc_env->me_psize * (omp->mp_pages - 1); + my->mc_over[toggle] = (char *)omp + my->mc_env->me_psize; + rc = mdb_env_cthr_toggle(my, 1); + if (rc) + goto done; + toggle = my->mc_toggle; + } + memcpy(NODEDATA(ni), &mo->mp_pgno, sizeof(pgno_t)); + } else if (ni->mn_flags & F_SUBDATA) { + MDB_db db; + + /* Need writable leaf */ + if (mp != leaf) { + mc.mc_pg[mc.mc_top] = leaf; + mdb_page_copy(leaf, mp, my->mc_env->me_psize); + mp = leaf; + ni = NODEPTR(mp, i); + } + + memcpy(&db, NODEDATA(ni), sizeof(db)); + my->mc_toggle = toggle; + rc = mdb_env_cwalk(my, &db.md_root, ni->mn_flags & F_DUPDATA); + if (rc) + goto done; + toggle = my->mc_toggle; + memcpy(NODEDATA(ni), &db, sizeof(db)); + } + } + } + } else { + mc.mc_ki[mc.mc_top]++; + if (mc.mc_ki[mc.mc_top] < n) { + pgno_t pg; +again: + ni = NODEPTR(mp, mc.mc_ki[mc.mc_top]); + pg = NODEPGNO(ni); + rc = mdb_page_get(txn, pg, &mp, NULL); + if (rc) + goto done; + mc.mc_top++; + mc.mc_snum++; + mc.mc_ki[mc.mc_top] = 0; + if (IS_BRANCH(mp)) { + /* Whenever we advance to a sibling branch page, + * we must proceed all the way down to its first leaf. + */ + mdb_page_copy(mc.mc_pg[mc.mc_top], mp, my->mc_env->me_psize); + goto again; + } else + mc.mc_pg[mc.mc_top] = mp; + continue; + } + } + if (my->mc_wlen[toggle] >= MDB_WBUF) { + rc = mdb_env_cthr_toggle(my, 1); + if (rc) + goto done; + toggle = my->mc_toggle; + } + mo = (MDB_page *)(my->mc_wbuf[toggle] + my->mc_wlen[toggle]); + mdb_page_copy(mo, mp, my->mc_env->me_psize); + mo->mp_pgno = my->mc_next_pgno++; + my->mc_wlen[toggle] += my->mc_env->me_psize; + if (mc.mc_top) { + /* Update parent if there is one */ + ni = NODEPTR(mc.mc_pg[mc.mc_top-1], mc.mc_ki[mc.mc_top-1]); + SETPGNO(ni, mo->mp_pgno); + mdb_cursor_pop(&mc); + } else { + /* Otherwise we're done */ + *pg = mo->mp_pgno; + break; + } + } +done: + free(buf); + return rc; +} + + /** Copy environment with compaction. */ +static int +mdb_env_copyfd1(MDB_env *env, HANDLE fd) +{ + MDB_meta *mm; + MDB_page *mp; + mdb_copy my; + MDB_txn *txn = NULL; + pthread_t thr; + int rc; + +#ifdef _WIN32 + my.mc_mutex = CreateMutex(NULL, FALSE, NULL); + my.mc_cond = CreateEvent(NULL, FALSE, FALSE, NULL); + my.mc_wbuf[0] = _aligned_malloc(MDB_WBUF*2, env->me_psize); + if (my.mc_wbuf[0] == NULL) + return errno; +#else + pthread_mutex_init(&my.mc_mutex, NULL); + pthread_cond_init(&my.mc_cond, NULL); + rc = posix_memalign((void **)&my.mc_wbuf[0], env->me_psize, MDB_WBUF*2); + if (rc) + return rc; +#endif + my.mc_wbuf[1] = my.mc_wbuf[0] + MDB_WBUF; + my.mc_wlen[0] = 0; + my.mc_wlen[1] = 0; + my.mc_olen[0] = 0; + my.mc_olen[1] = 0; + my.mc_next_pgno = 2; + my.mc_status = 0; + my.mc_new = 1; + my.mc_toggle = 0; + my.mc_env = env; + my.mc_fd = fd; + + /* Do the lock/unlock of the reader mutex before starting the + * write txn. Otherwise other read txns could block writers. + */ + rc = mdb_txn_begin(env, NULL, MDB_RDONLY, &txn); + if (rc) + return rc; + + if (env->me_txns) { + /* We must start the actual read txn after blocking writers */ + mdb_txn_reset0(txn, "reset-stage1"); + + /* Temporarily block writers until we snapshot the meta pages */ + LOCK_MUTEX_W(env); + + rc = mdb_txn_renew0(txn); + if (rc) { + UNLOCK_MUTEX_W(env); + goto leave; + } + } + + THREAD_CREATE(thr, mdb_env_copythr, &my); + mp = (MDB_page *)my.mc_wbuf[0]; + memset(mp, 0, 2*env->me_psize); + mp->mp_pgno = 0; + mp->mp_flags = P_META; + mm = (MDB_meta *)METADATA(mp); + mdb_env_init_meta0(env, mm); + mm->mm_address = env->me_metas[0]->mm_address; + + mp = (MDB_page *)(my.mc_wbuf[0] + env->me_psize); + mp->mp_pgno = 1; + mp->mp_flags = P_META; + *(MDB_meta *)METADATA(mp) = *mm; + mm = (MDB_meta *)METADATA(mp); + + /* Count the number of free pages, subtract from lastpg to find + * number of active pages + */ + { + MDB_ID freecount = 0; + MDB_cursor mc; + MDB_val key, data; + 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; + freecount += txn->mt_dbs[0].md_branch_pages + + txn->mt_dbs[0].md_leaf_pages + + txn->mt_dbs[0].md_overflow_pages; + + /* Set metapage 1 */ + mm->mm_last_pg = txn->mt_next_pgno - freecount - 1; + mm->mm_dbs[1] = txn->mt_dbs[1]; + mm->mm_dbs[1].md_root = mm->mm_last_pg; + mm->mm_txnid = 1; + } + my.mc_wlen[0] = env->me_psize * 2; + my.mc_txn = txn; + pthread_mutex_lock(&my.mc_mutex); + while(my.mc_new) + pthread_cond_wait(&my.mc_cond, &my.mc_mutex); + pthread_mutex_unlock(&my.mc_mutex); + rc = mdb_env_cwalk(&my, &txn->mt_dbs[1].md_root, 0); + if (rc == MDB_SUCCESS && my.mc_wlen[my.mc_toggle]) + rc = mdb_env_cthr_toggle(&my, 1); + mdb_env_cthr_toggle(&my, -1); + pthread_mutex_lock(&my.mc_mutex); + while(my.mc_new) + pthread_cond_wait(&my.mc_cond, &my.mc_mutex); + pthread_mutex_unlock(&my.mc_mutex); + THREAD_FINISH(thr); +leave: + mdb_txn_abort(txn); +#ifdef _WIN32 + CloseHandle(my.mc_cond); + CloseHandle(my.mc_mutex); + _aligned_free(my.mc_wbuf[0]); +#else + pthread_cond_destroy(&my.mc_cond); + pthread_mutex_destroy(&my.mc_mutex); + free(my.mc_wbuf[0]); +#endif + return rc; +} + + /** Copy environment as-is. */ +static int +mdb_env_copyfd0(MDB_env *env, HANDLE fd) +{ + MDB_txn *txn = NULL; + int rc; + size_t wsize; + char *ptr; +#ifdef _WIN32 + DWORD len, w2; +#define DO_WRITE(rc, fd, ptr, w2, len) rc = WriteFile(fd, ptr, w2, &len, NULL) +#else + ssize_t len; + size_t w2; +#define DO_WRITE(rc, fd, ptr, w2, len) len = write(fd, ptr, w2); rc = (len >= 0) +#endif + + /* Do the lock/unlock of the reader mutex before starting the + * write txn. Otherwise other read txns could block writers. + */ + rc = mdb_txn_begin(env, NULL, MDB_RDONLY, &txn); + if (rc) + return rc; + + if (env->me_txns) { + /* We must start the actual read txn after blocking writers */ + mdb_txn_reset0(txn, "reset-stage1"); + + /* Temporarily block writers until we snapshot the meta pages */ + LOCK_MUTEX_W(env); + + rc = mdb_txn_renew0(txn); + if (rc) { + UNLOCK_MUTEX_W(env); + goto leave; + } + } + + wsize = env->me_psize * 2; + ptr = env->me_map; + w2 = wsize; + while (w2 > 0) { + DO_WRITE(rc, fd, ptr, w2, len); + if (!rc) { + rc = ErrCode(); + break; + } else if (len > 0) { + rc = MDB_SUCCESS; + ptr += len; + w2 -= len; + continue; + } else { + /* Non-blocking or async handles are not supported */ + rc = EIO; + break; + } + } + if (env->me_txns) + UNLOCK_MUTEX_W(env); + + if (rc) + goto leave; + + w2 = txn->mt_next_pgno * env->me_psize; +#ifdef WIN32 + { + LARGE_INTEGER fsize; + GetFileSizeEx(env->me_fd, &fsize); + if (w2 > fsize.QuadPart) + w2 = fsize.QuadPart; + } +#else + { + struct stat st; + fstat(env->me_fd, &st); + if (w2 > (size_t)st.st_size) + w2 = st.st_size; + } +#endif + wsize = w2 - wsize; + while (wsize > 0) { + if (wsize > MAX_WRITE) + w2 = MAX_WRITE; + else + w2 = wsize; + DO_WRITE(rc, fd, ptr, w2, len); + if (!rc) { + rc = ErrCode(); + break; + } else if (len > 0) { + rc = MDB_SUCCESS; + ptr += len; + wsize -= len; + continue; + } else { + rc = EIO; + break; + } + } + +leave: + mdb_txn_abort(txn); + return rc; +} + +int +mdb_env_copyfd2(MDB_env *env, HANDLE fd, unsigned int flags) +{ + if (flags & MDB_CP_COMPACT) + return mdb_env_copyfd1(env, fd); + else + return mdb_env_copyfd0(env, fd); +} + +int +mdb_env_copyfd(MDB_env *env, HANDLE fd) +{ + return mdb_env_copyfd2(env, fd, 0); +} + +int +mdb_env_copy2(MDB_env *env, const char *path, unsigned int flags) +{ + int rc, len; + char *lpath; + HANDLE newfd = INVALID_HANDLE_VALUE; + + if (env->me_flags & MDB_NOSUBDIR) { + lpath = (char *)path; + } else { + len = strlen(path); + len += sizeof(DATANAME); + lpath = malloc(len); + if (!lpath) + return ENOMEM; + sprintf(lpath, "%s" DATANAME, path); + } + + /* The destination path must exist, but the destination file must not. + * We don't want the OS to cache the writes, since the source data is + * already in the OS cache. + */ +#ifdef _WIN32 + newfd = CreateFile(lpath, GENERIC_WRITE, 0, NULL, CREATE_NEW, + FILE_FLAG_NO_BUFFERING|FILE_FLAG_WRITE_THROUGH, NULL); +#else + newfd = open(lpath, O_WRONLY|O_CREAT|O_EXCL, 0666); +#endif + if (newfd == INVALID_HANDLE_VALUE) { + rc = ErrCode(); + goto leave; + } + +#ifdef O_DIRECT + /* Set O_DIRECT if the file system supports it */ + if ((rc = fcntl(newfd, F_GETFL)) != -1) + (void) fcntl(newfd, F_SETFL, rc | O_DIRECT); +#endif +#ifdef F_NOCACHE /* __APPLE__ */ + rc = fcntl(newfd, F_NOCACHE, 1); + if (rc) { + rc = ErrCode(); + goto leave; + } +#endif + + rc = mdb_env_copyfd2(env, newfd, flags); + +leave: + if (!(env->me_flags & MDB_NOSUBDIR)) + free(lpath); + if (newfd != INVALID_HANDLE_VALUE) + if (close(newfd) < 0 && rc == MDB_SUCCESS) + rc = ErrCode(); + + return rc; +} + +int +mdb_env_copy(MDB_env *env, const char *path) +{ + return mdb_env_copy2(env, path, 0); +} + int mdb_env_set_flags(MDB_env *env, unsigned int flag, int onoff) { @@ -7885,6 +8623,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) { @@ -7945,7 +8709,7 @@ mdb_env_info(MDB_env *env, MDB_envinfo *arg) return EINVAL; toggle = mdb_env_pick_meta(env); - arg->me_mapaddr = (env->me_flags & MDB_FIXEDMAP) ? env->me_map : 0; + arg->me_mapaddr = env->me_metas[toggle]->mm_address; arg->me_mapsize = env->me_mapsize; arg->me_maxreaders = env->me_maxreaders; @@ -8086,9 +8850,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; @@ -8113,7 +8880,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; @@ -8153,22 +8920,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. */ @@ -8195,6 +8964,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; } @@ -8206,7 +8978,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)) @@ -8225,10 +8997,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 */ @@ -8249,7 +9023,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; @@ -8258,7 +9032,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; @@ -8267,7 +9041,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; @@ -8276,7 +9050,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;