/** @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-2015 Howard Chu, Symas Corp.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
#ifndef _GNU_SOURCE
#define _GNU_SOURCE 1
#endif
-#include <sys/types.h>
-#include <sys/stat.h>
#ifdef _WIN32
+#include <malloc.h>
#include <windows.h>
/** getpid() returns int; MinGW defines pid_t but MinGW64 typedefs it
* as int64 which is wrong. MSVC doesn't define it at all, so just
* don't use it.
*/
#define MDB_PID_T int
+#define MDB_THR_T DWORD
+#include <sys/types.h>
+#include <sys/stat.h>
#ifdef __GNUC__
# include <sys/param.h>
#else
# endif
#endif
#else
+#include <sys/types.h>
+#include <sys/stat.h>
#define MDB_PID_T pid_t
+#define MDB_THR_T pthread_t
#include <sys/param.h>
#include <sys/uio.h>
#include <sys/mman.h>
#include <fcntl.h>
#endif
+#if defined(__mips) && defined(__linux)
+/* MIPS has cache coherency issues, requires explicit cache control */
+#include <asm/cachectl.h>
+extern int cacheflush(char *addr, int nbytes, int cache);
+#define CACHEFLUSH(addr, bytes, cache) cacheflush(addr, bytes, cache)
+#else
+#define CACHEFLUSH(addr, bytes, cache)
+#endif
+
+#if defined(__linux) && !defined(MDB_FDATASYNC_WORKS)
+/** fdatasync is broken on ext3/ext4fs on older kernels, see
+ * description in #mdb_env_open2 comments. You can safely
+ * define MDB_FDATASYNC_WORKS if this code will only be run
+ * on kernels 3.6 and newer.
+ */
+#define BROKEN_FDATASYNC
+#endif
+
#include <errno.h>
#include <limits.h>
#include <stddef.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
+
+#ifdef _MSC_VER
+#include <io.h>
+typedef SSIZE_T ssize_t;
+#else
#include <unistd.h>
+#endif
+
+#if defined(__sun) || defined(ANDROID)
+/* Most platforms have posix_memalign, older may only have memalign */
+#define HAVE_MEMALIGN 1
+#include <malloc.h>
+#endif
#if !(defined(BYTE_ORDER) || defined(__BYTE_ORDER))
#include <netinet/in.h>
#ifdef MDB_USE_POSIX_SEM
# define MDB_USE_HASH 1
#include <semaphore.h>
+#else
+#define MDB_USE_POSIX_MUTEX 1
#endif
#endif
+#if defined(_WIN32) + defined(MDB_USE_POSIX_SEM) \
+ + defined(MDB_USE_POSIX_MUTEX) != 1
+# error "Ambiguous shared-lock implementation"
+#endif
+
#ifdef USE_VALGRIND
#include <valgrind/memcheck.h>
#define VGMEMP_CREATE(h,r,z) VALGRIND_CREATE_MEMPOOL(h,r,z)
# error "Two's complement, reasonably sized integer types, please"
#endif
-/** @defgroup internal MDB Internals
+#ifdef __GNUC__
+/** Put infrequently used env functions in separate section */
+# ifdef __APPLE__
+# define ESECT __attribute__ ((section("__TEXT,text_env")))
+# else
+# define ESECT __attribute__ ((section("text_env")))
+# endif
+#else
+#define ESECT
+#endif
+
+#ifdef _MSC_VER
+#define CALL_CONV WINAPI
+#else
+#define CALL_CONV
+#endif
+
+/** @defgroup internal LMDB Internals
* @{
*/
/** @defgroup compat Compatibility Macros
* @{
*/
+ /** Features under development */
+#ifndef MDB_DEVEL
+#define MDB_DEVEL 0
+#endif
+
/** Wrapper around __func__, which is a C99 feature */
#if __STDC_VERSION__ >= 199901L
# define mdb_func_ __func__
# define mdb_func_ "<mdb_unknown>"
#endif
+/* Internal error codes, not exposed outside liblmdb */
+#define MDB_NO_ROOT (MDB_LAST_ERRCODE + 10)
+#ifdef _WIN32
+#define MDB_OWNERDEAD ((int) WAIT_ABANDONED)
+#elif defined(MDB_USE_POSIX_MUTEX) && defined(EOWNERDEAD)
+#define MDB_OWNERDEAD EOWNERDEAD /**< #LOCK_MUTEX0() result if dead owner */
+#endif
+
+#ifdef MDB_OWNERDEAD
+#define MDB_ROBUST_SUPPORTED 1
+#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
+typedef HANDLE mdb_mutex_t, mdb_mutexref_t;
#define pthread_key_t DWORD
#define pthread_self() GetCurrentThreadId()
#define pthread_key_create(x,y) \
#define pthread_key_delete(x) TlsFree(x)
#define pthread_getspecific(x) TlsGetValue(x)
#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 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)
-#define UNLOCK_MUTEX_W(env) pthread_mutex_unlock((env)->me_wmutex)
+#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) do{SignalObjectAndWait(*mutex, *cond, INFINITE, FALSE); WaitForSingleObject(*mutex, INFINITE);}while(0)
+#define THREAD_CREATE(thr,start,arg) thr=CreateThread(NULL,0,start,arg,0,NULL)
+#define THREAD_FINISH(thr) WaitForSingleObject(thr, INFINITE)
+#define LOCK_MUTEX0(mutex) WaitForSingleObject(mutex, INFINITE)
+#define UNLOCK_MUTEX(mutex) ReleaseMutex(mutex)
+#define mdb_mutex_consistent(mutex) 0
#define getpid() GetCurrentProcessId()
#define MDB_FDATASYNC(fd) (!FlushFileBuffers(fd))
#define MDB_MSYNC(addr,len,flags) (!FlushViewOfFile(addr,len))
#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 */
#ifdef MDB_USE_POSIX_SEM
-#define LOCK_MUTEX_R(env) mdb_sem_wait((env)->me_rmutex)
-#define UNLOCK_MUTEX_R(env) sem_post((env)->me_rmutex)
-#define LOCK_MUTEX_W(env) mdb_sem_wait((env)->me_wmutex)
-#define UNLOCK_MUTEX_W(env) sem_post((env)->me_wmutex)
+typedef sem_t *mdb_mutex_t, *mdb_mutexref_t;
+#define LOCK_MUTEX0(mutex) mdb_sem_wait(mutex)
+#define UNLOCK_MUTEX(mutex) sem_post(mutex)
static int
mdb_sem_wait(sem_t *sem)
return rc;
}
-#else
- /** Lock the reader mutex.
+#else /* MDB_USE_POSIX_MUTEX: */
+ /** Shared mutex/semaphore as it is stored (mdb_mutex_t), and as
+ * local variables keep it (mdb_mutexref_t).
+ *
+ * When #mdb_mutexref_t is a pointer declaration and #mdb_mutex_t is
+ * not, then it is array[size 1] so it can be assigned to a pointer.
+ * @{
*/
-#define LOCK_MUTEX_R(env) pthread_mutex_lock(&(env)->me_txns->mti_mutex)
- /** Unlock the reader mutex.
+typedef pthread_mutex_t mdb_mutex_t[1], *mdb_mutexref_t;
+ /* @} */
+ /** Lock the reader or writer mutex.
+ * Returns 0 or a code to give #mdb_mutex_failed(), as in #LOCK_MUTEX().
*/
-#define UNLOCK_MUTEX_R(env) pthread_mutex_unlock(&(env)->me_txns->mti_mutex)
-
- /** Lock the writer mutex.
- * Only a single write transaction is allowed at a time. Other writers
- * will block waiting for this mutex.
+#define LOCK_MUTEX0(mutex) pthread_mutex_lock(mutex)
+ /** Unlock the reader or writer mutex.
*/
-#define LOCK_MUTEX_W(env) pthread_mutex_lock(&(env)->me_txns->mti_wmutex)
- /** Unlock the writer mutex.
+#define UNLOCK_MUTEX(mutex) pthread_mutex_unlock(mutex)
+ /** Mark mutex-protected data as repaired, after death of previous owner.
*/
-#define UNLOCK_MUTEX_W(env) pthread_mutex_unlock(&(env)->me_txns->mti_wmutex)
+#define mdb_mutex_consistent(mutex) pthread_mutex_consistent(mutex)
#endif /* MDB_USE_POSIX_SEM */
/** Get the error code for the last failed system function.
/** @} */
+#ifdef MDB_ROBUST_SUPPORTED
+ /** Lock mutex, handle any error, set rc = result.
+ * Return 0 on success, nonzero (not rc) on error.
+ */
+#define LOCK_MUTEX(rc, env, mutex) \
+ (((rc) = LOCK_MUTEX0(mutex)) && \
+ ((rc) = mdb_mutex_failed(env, mutex, rc)))
+static int mdb_mutex_failed(MDB_env *env, mdb_mutexref_t mutex, int rc);
+#else
+#define LOCK_MUTEX(rc, env, mutex) ((rc) = LOCK_MUTEX0(mutex))
+#define mdb_mutex_failed(env, mutex, rc) (rc)
+#endif
+
#ifndef _WIN32
/** A flag for opening a file and requesting synchronous data writes.
* This is only used when writing a meta page. It's not strictly needed;
/** @brief The maximum size of a database page.
*
- * This is 32k, since it must fit in #MDB_page.%mp_upper.
+ * It is 32k or 64k, since value-PAGEBASE 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
* pressure from other processes is high. So until OSs have
* actual paging support for Huge pages, they're not viable.
*/
-#define MAX_PAGESIZE 0x8000
+#define MAX_PAGESIZE (PAGEBASE ? 0x10000 : 0x8000)
/** The minimum number of keys required in a database page.
* Setting this to a larger value will place a smaller bound on the
*/
#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.
*/
#define MDB_MAGIC 0xBEEFC0DE
/** The version number for a database's datafile format. */
-#define MDB_DATA_VERSION 1
+#define MDB_DATA_VERSION ((MDB_DEVEL) ? 999 : 1)
/** The version number for a database's lockfile format. */
#define MDB_LOCK_VERSION 1
- /** @brief The max size of a key we can write, or 0 for dynamic max.
+ /** @brief The max size of a key we can write, or 0 for computed max.
*
- * 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.
+ * This macro should normally be left alone or set to 0.
+ * Note that a database with big keys or dupsort data cannot be
+ * reliably modified by a liblmdb which uses a smaller max.
+ * The default is 511 for backwards compat, or 0 when #MDB_DEVEL.
+ *
+ * Other values are allowed, for backwards compat. However:
+ * A value bigger than the computed max can break if you do not
+ * know what you are doing, and liblmdb <= 0.9.10 can break when
+ * modifying a DB with keys/dupsort data bigger than its max.
*
* 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
+#define MDB_MAXKEYSIZE ((MDB_DEVEL) ? 0 : 511)
#endif
/** The maximum size of a key we can write to the environment. */
* started from so we can avoid overwriting any data used in that
* particular version.
*/
- txnid_t mrb_txnid;
+ volatile txnid_t mrb_txnid;
/** The process ID of the process owning this reader txn. */
- MDB_PID_T mrb_pid;
+ volatile MDB_PID_T mrb_pid;
/** The thread ID of the thread owning this txn. */
- pthread_t mrb_tid;
+ volatile MDB_THR_T mrb_tid;
} MDB_rxbody;
/** The actual reader record, with cacheline padding. */
* 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. */
char mtb_rmname[MNAME_LEN];
#else
/** Mutex protecting access to this table.
- * This is the reader lock that #LOCK_MUTEX_R acquires.
+ * This is the reader table lock used with LOCK_MUTEX().
*/
- pthread_mutex_t mtb_mutex;
+ mdb_mutex_t mtb_rmutex;
#endif
/** The ID of the last transaction committed to the database.
* This is recorded here only for convenience; the value can always
* be determined by reading the main database meta pages.
*/
- txnid_t mtb_txnid;
+ volatile txnid_t mtb_txnid;
/** The number of slots that have been used in the reader table.
* This always records the maximum count, it is not decremented
* when readers release their slots.
*/
- unsigned mtb_numreaders;
+ volatile unsigned mtb_numreaders;
} MDB_txbody;
/** The actual reader table definition. */
MDB_txbody mtb;
#define mti_magic mt1.mtb.mtb_magic
#define mti_format mt1.mtb.mtb_format
-#define mti_mutex mt1.mtb.mtb_mutex
+#define mti_rmutex mt1.mtb.mtb_rmutex
#define mti_rmname mt1.mtb.mtb_rmname
#define mti_txnid mt1.mtb.mtb_txnid
#define mti_numreaders mt1.mtb.mtb_numreaders
char mt2_wmname[MNAME_LEN];
#define mti_wmname mt2.mt2_wmname
#else
- pthread_mutex_t mt2_wmutex;
+ mdb_mutex_t mt2_wmutex;
#define mti_wmutex mt2.mt2_wmutex
#endif
char pad[(MNAME_LEN+CACHELINE-1) & ~(CACHELINE-1)];
#define mp_next mp_p.p_next
union {
pgno_t p_pgno; /**< page number */
- void * p_next; /**< for in-memory list of freed structs */
+ struct MDB_page *p_next; /**< for in-memory list of freed pages */
} mp_p;
uint16_t mp_pad;
/** @defgroup mdb_page Page Flags
#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 */
/** Address of first usable data byte in a page, after the header */
#define METADATA(p) ((void *)((char *)(p) + PAGEHDRSZ))
+ /** ITS#7713, change PAGEBASE to handle 65536 byte pages */
+#define PAGEBASE ((MDB_DEVEL) ? PAGEHDRSZ : 0)
+
/** Number of nodes on a page */
-#define NUMKEYS(p) (((p)->mp_lower - PAGEHDRSZ) >> 1)
+#define NUMKEYS(p) (((p)->mp_lower - (PAGEHDRSZ-PAGEBASE)) >> 1)
/** The amount of space remaining in the page */
#define SIZELEFT(p) (indx_t)((p)->mp_upper - (p)->mp_lower)
/** 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_pgs list */
+#define NEXT_LOOSE_PAGE(p) (*(MDB_page **)((p) + 2))
+
/** 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.
#define LEAFSIZE(k, d) (NODESIZE + (k)->mv_size + (d)->mv_size)
/** Address of node \b i in page \b p */
-#define NODEPTR(p, i) ((MDB_node *)((char *)(p) + (p)->mp_ptrs[i]))
+#define NODEPTR(p, i) ((MDB_node *)((char *)(p) + (p)->mp_ptrs[i] + PAGEBASE))
/** Address of the key for the node */
#define NODEKEY(node) (void *)((node)->mn_data)
#define FREE_DBI 0
/** Handle for the default DB. */
#define MAIN_DBI 1
+ /** Number of DBs in metapage (free and main) - also hardcoded elsewhere */
+#define CORE_DBS 2
+
+ /** Number of meta pages - also hardcoded elsewhere */
+#define NUM_METAS 2
/** Meta page content.
* A meta page is the start point for accessing a database snapshot.
* 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. */
+ /** Version number of this file. Must be set to #MDB_DATA_VERSION. */
uint32_t mm_version;
void *mm_address; /**< address for fixed mapping */
size_t mm_mapsize; /**< size of mmap region */
- MDB_db mm_dbs[2]; /**< first is free space, 2nd is main db */
+ MDB_db mm_dbs[CORE_DBS]; /**< first is free space, 2nd is main db */
/** The size of pages used in this DB */
-#define mm_psize mm_dbs[0].md_pad
+#define mm_psize mm_dbs[FREE_DBI].md_pad
/** Any persistent environment flags. @ref mdb_env */
-#define mm_flags mm_dbs[0].md_flags
+#define mm_flags mm_dbs[FREE_DBI].md_flags
pgno_t mm_last_pg; /**< last used page in file */
- txnid_t mm_txnid; /**< txnid that committed this page */
+ volatile txnid_t mm_txnid; /**< txnid that committed this page */
} MDB_meta;
/** Buffer for a stack-allocated meta page.
*/
struct MDB_txn {
MDB_txn *mt_parent; /**< parent of a nested txn */
- MDB_txn *mt_child; /**< nested txn under this txn */
+ /** Nested txn under this txn, set together with flag #MDB_TXN_HAS_CHILD */
+ MDB_txn *mt_child;
pgno_t mt_next_pgno; /**< next unallocated page */
/** The ID of this transaction. IDs are integers incrementing from 1.
* Only committed write transactions increment the ID. If a transaction
/** 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;
+ /* #Number of loose pages (#mt_loose_pgs) */
+ int mt_loose_count;
/** 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.
MDB_dbx *mt_dbxs;
/** Array of MDB_db records for each known DB */
MDB_db *mt_dbs;
+ /** Array of sequence numbers for each DB handle */
+ unsigned int *mt_dbiseqs;
/** @defgroup mt_dbflag Transaction DB Flags
* @ingroup internal
* @{
#define DB_STALE 0x02 /**< Named-DB record is older than txnID */
#define DB_NEW 0x04 /**< Named-DB handle opened in this txn */
#define DB_VALID 0x08 /**< DB handle is valid, see also #MDB_VALID */
+#define DB_USRVALID 0x10 /**< As #DB_VALID, but not set for #FREE_DBI */
/** @} */
/** In write txns, array of cursors for each DB */
MDB_cursor **mt_cursors;
/** Array of flags for each DB */
unsigned char *mt_dbflags;
- /** Number of DB records in use. This number only ever increments;
- * we don't decrement it when individual DB handles are closed.
+ /** Number of DB records in use, or 0 when the txn is finished.
+ * This number only ever increments until the txn finishes; we
+ * don't decrement it when individual DB handles are closed.
*/
MDB_dbi mt_numdbs;
* @ingroup internal
* @{
*/
-#define MDB_TXN_RDONLY 0x01 /**< read-only transaction */
-#define MDB_TXN_ERROR 0x02 /**< an error has occurred */
+ /** #mdb_txn_begin() flags */
+#define MDB_TXN_BEGIN_FLAGS MDB_RDONLY
+#define MDB_TXN_RDONLY MDB_RDONLY /**< read-only transaction */
+ /* internal txn flags */
+#define MDB_TXN_WRITEMAP MDB_WRITEMAP /**< copy of #MDB_env flag in writers */
+#define MDB_TXN_FINISHED 0x01 /**< txn is finished or never began */
+#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 */
+#define MDB_TXN_HAS_CHILD 0x10 /**< txn has an #MDB_txn.%mt_child */
+ /** most operations on the txn are currently illegal */
+#define MDB_TXN_BLOCKED (MDB_TXN_FINISHED|MDB_TXN_ERROR|MDB_TXN_HAS_CHILD)
/** @} */
unsigned int mt_flags; /**< @ref mdb_txn */
/** #dirty_list room: Array size - \#dirty pages visible to this txn.
#define MDB_ENV_ACTIVE 0x20000000U
/** me_txkey is set */
#define MDB_ENV_TXKEY 0x10000000U
+ /** fdatasync is unreliable */
+#define MDB_FSYNCONLY 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 */
unsigned int me_maxreaders; /**< size of the reader table */
- unsigned int me_numreaders; /**< max numreaders set by this env */
+ /** Max #MDB_txninfo.%mti_numreaders of interest to #mdb_env_close() */
+ volatile int me_close_readers;
MDB_dbi me_numdbs; /**< number of DBs opened */
MDB_dbi me_maxdbs; /**< size of the DB table */
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 */
+ MDB_meta *me_metas[NUM_METAS]; /**< pointers to the two meta pages */
void *me_pbuf; /**< scratch area for DUPSORT put() */
MDB_txn *me_txn; /**< current write transaction */
+ MDB_txn *me_txn0; /**< prealloc'd write transaction */
size_t me_mapsize; /**< size of the data memory map */
off_t me_size; /**< current file size */
pgno_t me_maxpg; /**< me_mapsize / me_psize */
MDB_dbx *me_dbxs; /**< array of static DB info */
uint16_t *me_dbflags; /**< array of flags from MDB_db.md_flags */
+ unsigned int *me_dbiseqs; /**< array of dbi sequence numbers */
pthread_key_t me_txkey; /**< thread-key for readers */
+ txnid_t me_pgoldest; /**< ID of oldest reader last time we looked */
MDB_pgstate me_pgstate; /**< state of old pages from freeDB */
# define me_pglast me_pgstate.mf_pglast
# define me_pghead me_pgstate.mf_pghead
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 */
- HANDLE me_wmutex;
-#elif defined(MDB_USE_POSIX_SEM)
- sem_t *me_rmutex; /* Shared mutexes are not supported */
- sem_t *me_wmutex;
+#endif
+#ifdef MDB_USE_POSIX_MUTEX /* Posix mutexes reside in shared mem */
+# define me_rmutex me_txns->mti_rmutex /**< Shared reader lock */
+# define me_wmutex me_txns->mti_wmutex /**< Shared writer lock */
+#else
+ mdb_mutex_t me_rmutex;
+ mdb_mutex_t me_wmutex;
#endif
void *me_userctx; /**< User-settable context */
MDB_assert_func *me_assert_func; /**< Callback for assertion failures */
/** 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, validity) \
+ ((txn) && (dbi)<(txn)->mt_numdbs && ((txn)->mt_dbflags[dbi] & (validity)))
+
+ /** Check for misused \b dbi handles */
+#define TXN_DBI_CHANGED(txn, dbi) \
+ ((txn)->mt_dbiseqs[dbi] != (txn)->mt_env->me_dbiseqs[dbi])
+
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);
+#define MDB_END_NAMES {"committed", "empty-commit", "abort", "reset", \
+ "reset-tmp", "fail-begin", "fail-beginchild"}
+enum {
+ /* mdb_txn_end operation number, for logging */
+ MDB_END_COMMITTED, MDB_END_EMPTY_COMMIT, MDB_END_ABORT, MDB_END_RESET,
+ MDB_END_RESET_TMP, MDB_END_FAIL_BEGIN, MDB_END_FAIL_BEGINCHILD
+};
+#define MDB_END_OPMASK 0x0F /**< mask for #mdb_txn_end() operation number */
+#define MDB_END_UPDATE 0x10 /**< update env state (DBIs) */
+#define MDB_END_FREE 0x20 /**< free txn unless it is #MDB_env.%me_txn0 */
+#define MDB_END_SLOT MDB_NOTLS /**< release any reader slot if #MDB_NOTLS */
+static void mdb_txn_end(MDB_txn *txn, unsigned mode);
+
static int mdb_page_get(MDB_txn *txn, pgno_t pgno, MDB_page **mp, int *lvl);
static int mdb_page_search_root(MDB_cursor *mc,
MDB_val *key, int modify);
pgno_t newpgno, unsigned int nflags);
static int mdb_env_read_header(MDB_env *env, MDB_meta *meta);
-static int mdb_env_pick_meta(const MDB_env *env);
+static MDB_meta *mdb_env_pick_meta(const MDB_env *env);
static int mdb_env_write_meta(MDB_txn *txn);
-#if !(defined(_WIN32) || defined(MDB_USE_POSIX_SEM)) /* Drop unused excl arg */
+#ifdef MDB_USE_POSIX_MUTEX /* Drop unused excl arg */
# define mdb_env_close0(env, excl) mdb_env_close1(env)
#endif
static void mdb_env_close0(MDB_env *env, int excl);
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);
static void mdb_cursor_init(MDB_cursor *mc, MDB_txn *txn, MDB_dbi dbi, MDB_xcursor *mx);
static void mdb_xcursor_init0(MDB_cursor *mc);
static void mdb_xcursor_init1(MDB_cursor *mc, MDB_node *node);
+static void mdb_xcursor_init2(MDB_cursor *mc, MDB_xcursor *src_mx, int force);
static int mdb_drop0(MDB_cursor *mc, int subs);
static void mdb_default_cmp(MDB_txn *txn, MDB_dbi dbi);
+static int mdb_reader_check0(MDB_env *env, int rlocked, int *dead);
/** @cond */
static MDB_cmp_func mdb_cmp_memn, mdb_cmp_memnr, mdb_cmp_int, mdb_cmp_cint, mdb_cmp_long;
/** @endcond */
+/** 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
+
#ifdef _WIN32
static SECURITY_DESCRIPTOR mdb_null_sd;
static SECURITY_ATTRIBUTES mdb_all_sa;
#endif
/** Return the library version info. */
-char *
+char * ESECT
mdb_version(int *major, int *minor, int *patch)
{
if (major) *major = MDB_VERSION_MAJOR;
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",
"MDB_PAGE_NOTFOUND: Requested page not found",
"MDB_CORRUPTED: Located page was wrong type",
- "MDB_PANIC: Update of meta page failed",
+ "MDB_PANIC: Update of meta page failed or environment had fatal error",
"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",
"MDB_MAP_RESIZED: Database contents grew beyond environment mapsize",
"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_TXN: Transaction must abort, has a child, or is invalid",
+ "MDB_BAD_VALSIZE: Unsupported size of key/DB name/data, or wrong DUPFIXED size",
+ "MDB_BAD_DBI: The specified DBI handle was closed/changed unexpectedly",
};
char *
mdb_strerror(int err)
{
+#ifdef _WIN32
+ /** HACK: pad 4KB on stack over the buf. Return system msgs in buf.
+ * This works as long as no function between the call to mdb_strerror
+ * and the actual use of the message uses more than 4K of stack.
+ */
+ char pad[4096];
+ char buf[1024], *ptr = buf;
+#endif
int i;
if (!err)
return ("Successful return: 0");
return mdb_errstr[i];
}
+#ifdef _WIN32
+ /* These are the C-runtime error codes we use. The comment indicates
+ * their numeric value, and the Win32 error they would correspond to
+ * if the error actually came from a Win32 API. A major mess, we should
+ * have used LMDB-specific error codes for everything.
+ */
+ switch(err) {
+ case ENOENT: /* 2, FILE_NOT_FOUND */
+ case EIO: /* 5, ACCESS_DENIED */
+ case ENOMEM: /* 12, INVALID_ACCESS */
+ case EACCES: /* 13, INVALID_DATA */
+ case EBUSY: /* 16, CURRENT_DIRECTORY */
+ case EINVAL: /* 22, BAD_COMMAND */
+ case ENOSPC: /* 28, OUT_OF_PAPER */
+ return strerror(err);
+ default:
+ ;
+ }
+ buf[0] = 0;
+ FormatMessage(FORMAT_MESSAGE_FROM_SYSTEM |
+ FORMAT_MESSAGE_IGNORE_INSERTS,
+ NULL, err, 0, ptr, sizeof(buf), (va_list *)pad);
+ return ptr;
+#else
return strerror(err);
+#endif
}
/** assert(3) variant in cursor context */
# define mdb_assert0(env, expr, expr_txt) ((expr) ? (void)0 : \
mdb_assert_fail(env, expr_txt, mdb_func_, __FILE__, __LINE__))
-static void
+static void ESECT
mdb_assert_fail(MDB_env *env, const char *expr_txt,
const char *func, const char *file, int line)
{
total = EVEN(total);
}
fprintf(stderr, "Total: header %d + contents %d + unused %d\n",
- IS_LEAF2(mp) ? PAGEHDRSZ : mp->mp_lower, total, SIZELEFT(mp));
+ IS_LEAF2(mp) ? PAGEHDRSZ : PAGEBASE + mp->mp_lower, total, SIZELEFT(mp));
}
void
/** Count all the pages in each DB and in the freelist
* and make sure it matches the actual number of pages
* being used.
+ * All named DBs must be open for a correct count.
*/
static void mdb_audit(MDB_txn *txn)
{
mdb_cursor_init(&mc, txn, FREE_DBI, NULL);
while ((rc = mdb_cursor_get(&mc, &key, &data, MDB_NEXT)) == 0)
freecount += *(MDB_ID *)data.mv_data;
+ mdb_tassert(txn, rc == MDB_NOTFOUND);
count = 0;
for (i = 0; i<txn->mt_numdbs; i++) {
MDB_xcursor mx;
+ if (!(txn->mt_dbflags[i] & DB_VALID))
+ continue;
mdb_cursor_init(&mc, txn, i, &mx);
if (txn->mt_dbs[i].md_root == P_INVALID)
continue;
txn->mt_dbs[i].md_leaf_pages +
txn->mt_dbs[i].md_overflow_pages;
if (txn->mt_dbs[i].md_flags & MDB_DUPSORT) {
- mdb_page_search(&mc, NULL, MDB_PS_FIRST);
- do {
+ rc = mdb_page_search(&mc, NULL, MDB_PS_FIRST);
+ for (; rc == MDB_SUCCESS; rc = mdb_cursor_sibling(&mc, 1)) {
unsigned j;
MDB_page *mp;
mp = mc.mc_pg[mc.mc_top];
}
}
}
- while (mdb_cursor_sibling(&mc, 1) == 0);
+ mdb_tassert(txn, rc == MDB_NOTFOUND);
}
}
- if (freecount + count + 2 /* metapages */ != txn->mt_next_pgno) {
+ if (freecount + count + NUM_METAS != txn->mt_next_pgno) {
fprintf(stderr, "audit: %lu freecount: %lu count: %lu total: %lu next_pgno: %lu\n",
- txn->mt_txnid, freecount, count+2, freecount+count+2, txn->mt_next_pgno);
+ txn->mt_txnid, freecount, count+NUM_METAS,
+ freecount+count+NUM_METAS, txn->mt_next_pgno);
}
}
#endif
int
mdb_dcmp(MDB_txn *txn, MDB_dbi dbi, const MDB_val *a, const MDB_val *b)
{
- return txn->mt_dbxs[dbi].md_dcmp(a, b);
+ MDB_cmp_func *dcmp = txn->mt_dbxs[dbi].md_dcmp;
+#if UINT_MAX < SIZE_MAX
+ if (dcmp == mdb_cmp_int && a->mv_size == sizeof(size_t))
+ dcmp = mdb_cmp_clong;
+#endif
+ return dcmp(a, b);
}
/** Allocate memory for a page.
}
return ret;
}
-
/** Free a single page.
* Saves single pages to a list, for future reuse.
* (This is not used for multi-page overflow pages.)
dl[0].mid = 0;
}
+/** Loosen 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;
+ MDB_txn *txn = mc->mc_txn;
+
+ if ((mp->mp_flags & P_DIRTY) && mc->mc_dbi != FREE_DBI) {
+ if (txn->mt_parent) {
+ MDB_ID2 *dl = 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);
+ 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) {
+ DPRINTF(("loosen db %d page %"Z"u", DDBI(mc),
+ mp->mp_pgno));
+ NEXT_LOOSE_PAGE(mp) = txn->mt_loose_pgs;
+ txn->mt_loose_pgs = mp;
+ txn->mt_loose_count++;
+ mp->mp_flags |= P_LOOSE;
+ } else {
+ int rc = mdb_midl_append(&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:
static int
mdb_pages_xkeep(MDB_cursor *mc, unsigned pflags, int all)
{
- enum { Mask = P_SUBP|P_DIRTY|P_KEEP };
+ enum { Mask = P_SUBP|P_DIRTY|P_LOOSE|P_KEEP };
MDB_txn *txn = mc->mc_txn;
MDB_cursor *m3;
MDB_xcursor *mx;
/* Estimate how much space this op will take */
i = m0->mc_db->md_depth;
/* Named DBs also dirty the main DB */
- if (m0->mc_dbi > MAIN_DBI)
+ if (m0->mc_dbi >= CORE_DBS)
i += txn->mt_dbs[MAIN_DBI].md_depth;
/* For puts, roughly factor in the key+data size */
if (key)
for (i=dl[0].mid; i && need; i--) {
MDB_ID pn = dl[i].mid << 1;
dp = dl[i].mptr;
- if (dp->mp_flags & P_KEEP)
+ if (dp->mp_flags & (P_LOOSE|P_KEEP))
continue;
/* Can't spill twice, make sure it's not already in a parent's
* spill list.
MDB_ID2 mid;
int rc, (*insert)(MDB_ID2L, MDB_ID2 *);
- if (txn->mt_env->me_flags & MDB_WRITEMAP) {
+ if (txn->mt_flags & MDB_TXN_WRITEMAP) {
insert = mdb_mid2l_append;
} else {
insert = mdb_mid2l_insert;
#else
enum { Paranoid = 0, Max_retries = INT_MAX /*infinite*/ };
#endif
- int rc, retry = Max_retries;
+ int rc, retry = num * 60;
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, n2 = num-1;
+ unsigned i, j, mop_len = mop ? mop[0] : 0, n2 = num-1;
MDB_page *np;
txnid_t oldest = 0, last;
MDB_cursor_op op;
MDB_cursor m2;
+ int found_old = 0;
+
+ /* 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);
+ txn->mt_loose_count--;
+ DPRINTF(("db %d use loose page %"Z"u", DDBI(mc),
+ np->mp_pgno));
+ *mp = np;
+ return MDB_SUCCESS;
+ }
*mp = NULL;
for (op = MDB_FIRST;; op = MDB_NEXT) {
MDB_val key, data;
MDB_node *leaf;
- pgno_t *idl, old_id, new_id;
+ pgno_t *idl;
/* Seek a big enough contiguous page range. Prefer
* pages at the tail, just truncating the list.
if (mop[i-n2] == pgno+n2)
goto search_done;
} while (--i > n2);
- if (Max_retries < INT_MAX && --retry < 0)
+ if (--retry < 0)
break;
}
if (op == MDB_FIRST) { /* 1st iteration */
/* Prepare to fetch more and coalesce */
- oldest = mdb_find_oldest(txn);
last = env->me_pglast;
+ oldest = env->me_pgoldest;
mdb_cursor_init(&m2, txn, FREE_DBI, NULL);
if (last) {
op = MDB_SET_RANGE;
last++;
/* Do not fetch more if the record will be too recent */
- if (oldest <= last)
- break;
+ if (oldest <= last) {
+ if (!found_old) {
+ oldest = mdb_find_oldest(txn);
+ env->me_pgoldest = oldest;
+ found_old = 1;
+ }
+ if (oldest <= last)
+ break;
+ }
rc = mdb_cursor_get(&m2, &key, NULL, op);
if (rc) {
if (rc == MDB_NOTFOUND)
goto fail;
}
last = *(txnid_t*)key.mv_data;
- if (oldest <= last)
- break;
+ if (oldest <= last) {
+ if (!found_old) {
+ oldest = mdb_find_oldest(txn);
+ env->me_pgoldest = oldest;
+ found_old = 1;
+ }
+ if (oldest <= last)
+ break;
+ }
np = m2.mc_pg[m2.mc_top];
leaf = NODEPTR(np, m2.mc_ki[m2.mc_top]);
if ((rc = mdb_node_read(txn, leaf, &data)) != MDB_SUCCESS)
#if (MDB_DEBUG) > 1
DPRINTF(("IDL read txn %"Z"u root %"Z"u num %u",
last, txn->mt_dbs[FREE_DBI].md_root, i));
- for (k = i; k; k--)
- DPRINTF(("IDL %"Z"u", idl[k]));
+ for (j = i; j; j--)
+ DPRINTF(("IDL %"Z"u", idl[j]));
#endif
/* Merge in descending sorted order */
- j = mop_len;
- k = mop_len += i;
- mop[0] = (pgno_t)-1;
- old_id = mop[j];
- while (i) {
- new_id = idl[i--];
- for (; old_id < new_id; old_id = mop[--j])
- mop[k--] = old_id;
- mop[k--] = new_id;
- }
- mop[0] = mop_len;
+ mdb_midl_xmerge(mop, idl);
+ mop_len = mop[0];
}
/* Use new pages from the map when nothing suitable in the freeDB */
* alignment so memcpy may copy words instead of bytes.
*/
if ((unused &= -Align) && !IS_LEAF2(src)) {
- upper &= -Align;
- memcpy(dst, src, (lower + (Align-1)) & -Align);
+ upper = (upper + PAGEBASE) & -Align;
+ memcpy(dst, src, (lower + PAGEBASE + (Align-1)) & -Align);
memcpy((pgno_t *)((char *)dst+upper), (pgno_t *)((char *)src+upper),
psize - upper);
} else {
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]);
mdb_env_sync(MDB_env *env, int force)
{
int rc = 0;
+ if (env->me_flags & MDB_RDONLY)
+ return EACCES;
if (force || !F_ISSET(env->me_flags, MDB_NOSYNC)) {
if (env->me_flags & MDB_WRITEMAP) {
int flags = ((env->me_flags & MDB_MAPASYNC) && !force)
rc = ErrCode();
#endif
} else {
+#ifdef BROKEN_FDATASYNC
+ if (env->me_flags & MDB_FSYNCONLY) {
+ if (fsync(env->me_fd))
+ rc = ErrCode();
+ } else
+#endif
if (MDB_FDATASYNC(env->me_fd))
rc = ErrCode();
}
}
}
-#if !(MDB_DEBUG)
-#define mdb_txn_reset0(txn, act) mdb_txn_reset0(txn)
-#endif
-static void
-mdb_txn_reset0(MDB_txn *txn, const char *act);
-
#if !(MDB_PIDLOCK) /* Currently the same as defined(_WIN32) */
enum Pidlock_op {
Pidset, Pidcheck
MDB_env *env = txn->mt_env;
MDB_txninfo *ti = env->me_txns;
MDB_meta *meta;
- unsigned int i, nr;
+ unsigned int i, nr, flags = txn->mt_flags;
uint16_t x;
int rc, new_notls = 0;
- /* Setup db info */
- txn->mt_numdbs = env->me_numdbs;
- txn->mt_dbxs = env->me_dbxs; /* mostly static anyway */
-
- if (txn->mt_flags & MDB_TXN_RDONLY) {
+ if ((flags &= MDB_TXN_RDONLY) != 0) {
if (!ti) {
- meta = env->me_metas[ mdb_env_pick_meta(env) ];
+ meta = mdb_env_pick_meta(env);
txn->mt_txnid = meta->mm_txnid;
txn->mt_u.reader = NULL;
} else {
return MDB_BAD_RSLOT;
} else {
MDB_PID_T pid = env->me_pid;
- pthread_t tid = pthread_self();
+ MDB_THR_T tid = pthread_self();
+ mdb_mutexref_t rmutex = env->me_rmutex;
if (!env->me_live_reader) {
rc = mdb_reader_pid(env, Pidset, pid);
env->me_live_reader = 1;
}
- LOCK_MUTEX_R(env);
+ if (LOCK_MUTEX(rc, env, rmutex))
+ return rc;
nr = ti->mti_numreaders;
for (i=0; i<nr; i++)
if (ti->mti_readers[i].mr_pid == 0)
break;
if (i == env->me_maxreaders) {
- UNLOCK_MUTEX_R(env);
+ UNLOCK_MUTEX(rmutex);
return MDB_READERS_FULL;
}
- ti->mti_readers[i].mr_pid = pid;
- ti->mti_readers[i].mr_tid = tid;
+ r = &ti->mti_readers[i];
+ /* Claim the reader slot, carefully since other code
+ * uses the reader table un-mutexed: First reset the
+ * slot, next publish it in mti_numreaders. After
+ * that, it is safe for mdb_env_close() to touch it.
+ * When it will be closed, we can finally claim it.
+ */
+ r->mr_pid = 0;
+ r->mr_txnid = (txnid_t)-1;
+ r->mr_tid = tid;
if (i == nr)
ti->mti_numreaders = ++nr;
- /* Save numreaders for un-mutexed mdb_env_close() */
- env->me_numreaders = nr;
- UNLOCK_MUTEX_R(env);
+ env->me_close_readers = nr;
+ r->mr_pid = pid;
+ UNLOCK_MUTEX(rmutex);
- r = &ti->mti_readers[i];
new_notls = (env->me_flags & MDB_NOTLS);
if (!new_notls && (rc=pthread_setspecific(env->me_txkey, r))) {
r->mr_pid = 0;
return rc;
}
}
- txn->mt_txnid = r->mr_txnid = ti->mti_txnid;
+ do /* LY: Retry on a race, ITS#7970. */
+ r->mr_txnid = ti->mti_txnid;
+ while(r->mr_txnid != ti->mti_txnid);
+ txn->mt_txnid = r->mr_txnid;
txn->mt_u.reader = r;
meta = env->me_metas[txn->mt_txnid & 1];
}
+
} else {
+ /* Not yet touching txn == env->me_txn0, it may be active */
if (ti) {
- LOCK_MUTEX_W(env);
-
+ if (LOCK_MUTEX(rc, env, env->me_wmutex))
+ return rc;
txn->mt_txnid = ti->mti_txnid;
meta = env->me_metas[txn->mt_txnid & 1];
} else {
- meta = env->me_metas[ mdb_env_pick_meta(env) ];
+ meta = mdb_env_pick_meta(env);
txn->mt_txnid = meta->mm_txnid;
}
txn->mt_txnid++;
if (txn->mt_txnid == mdb_debug_start)
mdb_debug = 1;
#endif
+ txn->mt_child = NULL;
+ txn->mt_loose_pgs = NULL;
+ txn->mt_loose_count = 0;
txn->mt_dirty_room = MDB_IDL_UM_MAX;
txn->mt_u.dirty_list = env->me_dirty_list;
txn->mt_u.dirty_list[0].mid = 0;
txn->mt_free_pgs[0] = 0;
txn->mt_spill_pgs = NULL;
env->me_txn = txn;
+ memcpy(txn->mt_dbiseqs, env->me_dbiseqs, env->me_maxdbs * sizeof(unsigned int));
}
/* Copy the DB info and flags */
- memcpy(txn->mt_dbs, meta->mm_dbs, 2 * sizeof(MDB_db));
+ memcpy(txn->mt_dbs, meta->mm_dbs, CORE_DBS * sizeof(MDB_db));
/* Moved to here to avoid a data race in read TXNs */
txn->mt_next_pgno = meta->mm_last_pg+1;
- for (i=2; i<txn->mt_numdbs; i++) {
+ txn->mt_flags = flags;
+
+ /* Setup db info */
+ txn->mt_numdbs = env->me_numdbs;
+ for (i=CORE_DBS; i<txn->mt_numdbs; i++) {
x = env->me_dbflags[i];
txn->mt_dbs[i].md_flags = x & PERSISTENT_FLAGS;
- txn->mt_dbflags[i] = (x & MDB_VALID) ? DB_VALID|DB_STALE : 0;
+ txn->mt_dbflags[i] = (x & MDB_VALID) ? DB_VALID|DB_USRVALID|DB_STALE : 0;
}
- txn->mt_dbflags[0] = txn->mt_dbflags[1] = DB_VALID;
+ txn->mt_dbflags[MAIN_DBI] = DB_VALID|DB_USRVALID;
+ txn->mt_dbflags[FREE_DBI] = DB_VALID;
- if (env->me_maxpg < txn->mt_next_pgno) {
- mdb_txn_reset0(txn, "renew0-mapfail");
- if (new_notls) {
- txn->mt_u.reader->mr_pid = 0;
- txn->mt_u.reader = NULL;
- }
- return MDB_MAP_RESIZED;
+ if (env->me_flags & MDB_FATAL_ERROR) {
+ DPUTS("environment had fatal error, must shutdown!");
+ rc = MDB_PANIC;
+ } else if (env->me_maxpg < txn->mt_next_pgno) {
+ rc = MDB_MAP_RESIZED;
+ } else {
+ return MDB_SUCCESS;
}
-
- return MDB_SUCCESS;
+ mdb_txn_end(txn, new_notls /*0 or MDB_END_SLOT*/ | MDB_END_FAIL_BEGIN);
+ return rc;
}
int
{
int rc;
- if (!txn || txn->mt_dbxs) /* A reset txn has mt_dbxs==NULL */
+ if (!txn || !F_ISSET(txn->mt_flags, MDB_TXN_RDONLY|MDB_TXN_FINISHED))
return EINVAL;
- if (txn->mt_env->me_flags & MDB_FATAL_ERROR) {
- DPUTS("environment had fatal error, must shutdown!");
- return MDB_PANIC;
- }
-
rc = mdb_txn_renew0(txn);
if (rc == MDB_SUCCESS) {
DPRINTF(("renew txn %"Z"u%c %p on mdbenv %p, root page %"Z"u",
{
MDB_txn *txn;
MDB_ntxn *ntxn;
- int rc, size, tsize = sizeof(MDB_txn);
+ int rc, size, tsize;
- if (env->me_flags & MDB_FATAL_ERROR) {
- DPUTS("environment had fatal error, must shutdown!");
- return MDB_PANIC;
- }
- if ((env->me_flags & MDB_RDONLY) && !(flags & MDB_RDONLY))
+ flags &= MDB_TXN_BEGIN_FLAGS;
+ flags |= env->me_flags & MDB_WRITEMAP;
+
+ if (env->me_flags & MDB_RDONLY & ~flags) /* write txn in RDONLY env */
return EACCES;
+
if (parent) {
/* Nested transactions: Max 1 child, write txns only, no writemap */
- if (parent->mt_child ||
- (flags & MDB_RDONLY) ||
- (parent->mt_flags & (MDB_TXN_RDONLY|MDB_TXN_ERROR)) ||
- (env->me_flags & MDB_WRITEMAP))
- {
+ flags |= parent->mt_flags;
+ if (flags & (MDB_RDONLY|MDB_WRITEMAP|MDB_TXN_BLOCKED)) {
return (parent->mt_flags & MDB_TXN_RDONLY) ? EINVAL : MDB_BAD_TXN;
}
- tsize = sizeof(MDB_ntxn);
+ /* Child txns save MDB_pgstate and use own copy of cursors */
+ size = env->me_maxdbs * (sizeof(MDB_db)+sizeof(MDB_cursor *)+1);
+ size += tsize = sizeof(MDB_ntxn);
+ } else if (flags & MDB_RDONLY) {
+ size = env->me_maxdbs * (sizeof(MDB_db)+1);
+ size += tsize = sizeof(MDB_txn);
+ } else {
+ /* Reuse preallocated write txn. However, do not touch it until
+ * mdb_txn_renew0() succeeds, since it currently may be active.
+ */
+ txn = env->me_txn0;
+ goto renew;
}
- size = tsize + env->me_maxdbs * (sizeof(MDB_db)+1);
- if (!(flags & MDB_RDONLY))
- size += env->me_maxdbs * sizeof(MDB_cursor *);
-
if ((txn = calloc(1, size)) == NULL) {
- DPRINTF(("calloc: %s", strerror(ErrCode())));
+ DPRINTF(("calloc: %s", strerror(errno)));
return ENOMEM;
}
+ txn->mt_dbxs = env->me_dbxs; /* static */
txn->mt_dbs = (MDB_db *) ((char *)txn + tsize);
- if (flags & MDB_RDONLY) {
- txn->mt_flags |= MDB_TXN_RDONLY;
- txn->mt_dbflags = (unsigned char *)(txn->mt_dbs + env->me_maxdbs);
- } else {
- txn->mt_cursors = (MDB_cursor **)(txn->mt_dbs + env->me_maxdbs);
- txn->mt_dbflags = (unsigned char *)(txn->mt_cursors + env->me_maxdbs);
- }
+ txn->mt_dbflags = (unsigned char *)txn + size - env->me_maxdbs;
+ txn->mt_flags = flags;
txn->mt_env = env;
if (parent) {
unsigned int i;
+ txn->mt_cursors = (MDB_cursor **)(txn->mt_dbs + env->me_maxdbs);
+ txn->mt_dbiseqs = parent->mt_dbiseqs;
txn->mt_u.dirty_list = malloc(sizeof(MDB_ID2)*MDB_IDL_UM_SIZE);
if (!txn->mt_u.dirty_list ||
!(txn->mt_free_pgs = mdb_midl_alloc(MDB_IDL_UM_MAX)))
txn->mt_u.dirty_list[0].mid = 0;
txn->mt_spill_pgs = NULL;
txn->mt_next_pgno = parent->mt_next_pgno;
+ parent->mt_flags |= MDB_TXN_HAS_CHILD;
parent->mt_child = txn;
txn->mt_parent = parent;
txn->mt_numdbs = parent->mt_numdbs;
- txn->mt_flags = parent->mt_flags;
- txn->mt_dbxs = parent->mt_dbxs;
memcpy(txn->mt_dbs, parent->mt_dbs, txn->mt_numdbs * sizeof(MDB_db));
/* Copy parent's mt_dbflags, but clear DB_NEW */
for (i=0; i<txn->mt_numdbs; i++)
if (!rc)
rc = mdb_cursor_shadow(parent, txn);
if (rc)
- mdb_txn_reset0(txn, "beginchild-fail");
- } else {
+ mdb_txn_end(txn, MDB_END_FAIL_BEGINCHILD);
+ } else { /* MDB_RDONLY */
+ txn->mt_dbiseqs = env->me_dbiseqs;
+renew:
rc = mdb_txn_renew0(txn);
}
- if (rc)
- free(txn);
- else {
+ if (rc) {
+ if (txn != env->me_txn0)
+ free(txn);
+ } else {
+ txn->mt_flags |= flags; /* could not change txn=me_txn0 earlier */
*ret = txn;
DPRINTF(("begin txn %"Z"u%c %p on mdbenv %p, root page %"Z"u",
- txn->mt_txnid, (txn->mt_flags & MDB_TXN_RDONLY) ? 'r' : 'w',
+ txn->mt_txnid, (flags & MDB_RDONLY) ? 'r' : 'w',
(void *) txn, (void *) env, txn->mt_dbs[MAIN_DBI].md_root));
}
return txn->mt_env;
}
+size_t
+mdb_txn_id(MDB_txn *txn)
+{
+ if(!txn) return 0;
+ return txn->mt_txnid;
+}
+
/** Export or close DBI handles opened in this txn. */
static void
mdb_dbis_update(MDB_txn *txn, int keep)
MDB_env *env = txn->mt_env;
unsigned char *tdbflags = txn->mt_dbflags;
- for (i = n; --i >= 2;) {
+ for (i = n; --i >= CORE_DBS;) {
if (tdbflags[i] & DB_NEW) {
if (keep) {
env->me_dbflags[i] = txn->mt_dbs[i].md_flags | MDB_VALID;
} else {
char *ptr = env->me_dbxs[i].md_name.mv_data;
- env->me_dbxs[i].md_name.mv_data = NULL;
- env->me_dbxs[i].md_name.mv_size = 0;
- env->me_dbflags[i] = 0;
- free(ptr);
+ if (ptr) {
+ env->me_dbxs[i].md_name.mv_data = NULL;
+ env->me_dbxs[i].md_name.mv_size = 0;
+ env->me_dbflags[i] = 0;
+ env->me_dbiseqs[i]++;
+ free(ptr);
+ }
}
}
}
env->me_numdbs = n;
}
-/** Common code for #mdb_txn_reset() and #mdb_txn_abort().
+/** End a transaction, except successful commit of a nested transaction.
* May be called twice for readonly txns: First reset it, then abort.
- * @param[in] txn the transaction handle to reset
- * @param[in] act why the transaction is being reset
+ * @param[in] txn the transaction handle to end
+ * @param[in] mode why and how to end the transaction
*/
static void
-mdb_txn_reset0(MDB_txn *txn, const char *act)
+mdb_txn_end(MDB_txn *txn, unsigned mode)
{
MDB_env *env = txn->mt_env;
+#if MDB_DEBUG
+ static const char *const names[] = MDB_END_NAMES;
+#endif
- /* Close any DBI handles opened in this txn */
- mdb_dbis_update(txn, 0);
+ /* Export or close DBI handles opened in this txn */
+ mdb_dbis_update(txn, mode & MDB_END_UPDATE);
DPRINTF(("%s txn %"Z"u%c %p on mdbenv %p, root page %"Z"u",
- act, txn->mt_txnid, (txn->mt_flags & MDB_TXN_RDONLY) ? 'r' : 'w',
+ names[mode & MDB_END_OPMASK],
+ txn->mt_txnid, (txn->mt_flags & MDB_TXN_RDONLY) ? 'r' : 'w',
(void *) txn, (void *)env, txn->mt_dbs[MAIN_DBI].md_root));
if (F_ISSET(txn->mt_flags, MDB_TXN_RDONLY)) {
if (txn->mt_u.reader) {
txn->mt_u.reader->mr_txnid = (txnid_t)-1;
- if (!(env->me_flags & MDB_NOTLS))
+ if (!(env->me_flags & MDB_NOTLS)) {
txn->mt_u.reader = NULL; /* txn does not own reader */
+ } else if (mode & MDB_END_SLOT) {
+ txn->mt_u.reader->mr_pid = 0;
+ txn->mt_u.reader = NULL;
+ } /* else txn owns the slot until it does MDB_END_SLOT */
}
- txn->mt_numdbs = 0; /* close nothing if called again */
- txn->mt_dbxs = NULL; /* mark txn as reset */
- } else {
- mdb_cursors_close(txn, 0);
+ txn->mt_numdbs = 0; /* prevent further DBI activity */
+ txn->mt_flags |= MDB_TXN_FINISHED;
+
+ } else if (!F_ISSET(txn->mt_flags, MDB_TXN_FINISHED)) {
+ pgno_t *pghead = env->me_pghead;
+ if (!(mode & MDB_END_UPDATE)) /* !(already closed cursors) */
+ mdb_cursors_close(txn, 0);
if (!(env->me_flags & MDB_WRITEMAP)) {
mdb_dlist_free(txn);
}
- mdb_midl_free(env->me_pghead);
- if (txn->mt_parent) {
+ txn->mt_numdbs = 0;
+ txn->mt_flags = MDB_TXN_FINISHED;
+
+ if (!txn->mt_parent) {
+ mdb_midl_shrink(&txn->mt_free_pgs);
+ env->me_free_pgs = txn->mt_free_pgs;
+ /* me_pgstate: */
+ env->me_pghead = NULL;
+ env->me_pglast = 0;
+
+ env->me_txn = NULL;
+ mode = 0; /* txn == env->me_txn0, do not free() it */
+
+ /* The writer mutex was locked in mdb_txn_begin. */
+ if (env->me_txns)
+ UNLOCK_MUTEX(env->me_wmutex);
+ } else {
txn->mt_parent->mt_child = NULL;
+ txn->mt_parent->mt_flags &= ~MDB_TXN_HAS_CHILD;
env->me_pgstate = ((MDB_ntxn *)txn)->mnt_pgstate;
mdb_midl_free(txn->mt_free_pgs);
mdb_midl_free(txn->mt_spill_pgs);
free(txn->mt_u.dirty_list);
- return;
}
- if (mdb_midl_shrink(&txn->mt_free_pgs))
- env->me_free_pgs = txn->mt_free_pgs;
- env->me_pghead = NULL;
- env->me_pglast = 0;
-
- env->me_txn = NULL;
- /* The writer mutex was locked in mdb_txn_begin. */
- if (env->me_txns)
- UNLOCK_MUTEX_W(env);
+ mdb_midl_free(pghead);
}
+
+ if (mode & MDB_END_FREE)
+ free(txn);
}
void
if (!(txn->mt_flags & MDB_TXN_RDONLY))
return;
- mdb_txn_reset0(txn, "reset");
+ mdb_txn_end(txn, MDB_END_RESET);
}
void
if (txn->mt_child)
mdb_txn_abort(txn->mt_child);
- mdb_txn_reset0(txn, "abort");
- /* Free reader slot tied to this txn (if MDB_NOTLS && writable FS) */
- if ((txn->mt_flags & MDB_TXN_RDONLY) && txn->mt_u.reader)
- txn->mt_u.reader->mr_pid = 0;
-
- free(txn);
+ mdb_txn_end(txn, MDB_END_ABORT|MDB_END_SLOT|MDB_END_FREE);
}
/** Save the freelist as of this transaction to the freeDB.
return rc;
}
+ if (!env->me_pghead && txn->mt_loose_pgs) {
+ /* Put loose page numbers in mt_free_pgs, since
+ * we may be unable to return them to me_pghead.
+ */
+ MDB_page *mp = txn->mt_loose_pgs;
+ if ((rc = mdb_midl_need(&txn->mt_free_pgs, txn->mt_loose_count)) != 0)
+ return rc;
+ for (; mp; mp = NEXT_LOOSE_PAGE(mp))
+ mdb_midl_xappend(txn->mt_free_pgs, mp->mp_pgno);
+ txn->mt_loose_pgs = NULL;
+ txn->mt_loose_count = 0;
+ }
+
/* MDB_RESERVE cancels meminit in ovpage malloc (when no WRITEMAP) */
clean_limit = (env->me_flags & (MDB_NOMEMINIT|MDB_WRITEMAP))
? SSIZE_MAX : maxfree_1pg;
}
mop = env->me_pghead;
- mop_len = mop ? mop[0] : 0;
+ mop_len = (mop ? mop[0] : 0) + txn->mt_loose_count;
/* Reserve records for me_pghead[]. Split it if multi-page,
* to avoid searching freeDB for a page range. Use keys in
total_room += head_room;
}
+ /* Return loose page numbers to me_pghead, though usually none are
+ * left at this point. The pages themselves remain in dirty_list.
+ */
+ if (txn->mt_loose_pgs) {
+ MDB_page *mp = txn->mt_loose_pgs;
+ unsigned count = txn->mt_loose_count;
+ MDB_IDL loose;
+ /* Room for loose pages + temp IDL with same */
+ if ((rc = mdb_midl_need(&env->me_pghead, 2*count+1)) != 0)
+ return rc;
+ mop = env->me_pghead;
+ loose = mop + MDB_IDL_ALLOCLEN(mop) - count;
+ for (count = 0; mp; mp = NEXT_LOOSE_PAGE(mp))
+ loose[ ++count ] = mp->mp_pgno;
+ loose[0] = count;
+ mdb_midl_sort(loose);
+ mdb_midl_xmerge(mop, loose);
+ txn->mt_loose_pgs = NULL;
+ txn->mt_loose_count = 0;
+ mop_len = mop[0];
+ }
+
/* Fill in the reserved me_pghead records */
rc = MDB_SUCCESS;
if (mop_len) {
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;
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);
- /* Drop MDB_CURRENT when changing the data size */
- key.mv_data = &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;
while (++i <= pagecount) {
dp = dl[i].mptr;
/* Don't flush this page yet */
- if (dp->mp_flags & P_KEEP) {
- dp->mp_flags ^= P_KEEP;
+ if (dp->mp_flags & (P_LOOSE|P_KEEP)) {
+ dp->mp_flags &= ~P_KEEP;
dl[++j] = dl[i];
continue;
}
if (++i <= pagecount) {
dp = dl[i].mptr;
/* Don't flush this page yet */
- if (dp->mp_flags & P_KEEP) {
- dp->mp_flags ^= P_KEEP;
+ if (dp->mp_flags & (P_LOOSE|P_KEEP)) {
+ dp->mp_flags &= ~P_KEEP;
dl[i].mid = 0;
continue;
}
/* Write up to MDB_COMMIT_PAGES dirty pages at a time. */
if (pos!=next_pos || n==MDB_COMMIT_PAGES || wsize+size>MAX_WRITE) {
if (n) {
+retry_write:
/* Write previous page(s) */
#ifdef MDB_USE_PWRITEV
wres = pwritev(env->me_fd, iov, n, wpos);
if (n == 1) {
wres = pwrite(env->me_fd, iov[0].iov_base, wsize, wpos);
} else {
+retry_seek:
if (lseek(env->me_fd, wpos, SEEK_SET) == -1) {
rc = ErrCode();
+ if (rc == EINTR)
+ goto retry_seek;
DPRINTF(("lseek: %s", strerror(rc)));
return rc;
}
if (wres != wsize) {
if (wres < 0) {
rc = ErrCode();
+ if (rc == EINTR)
+ goto retry_write;
DPRINTF(("Write error: %s", strerror(rc)));
} else {
rc = EIO; /* TODO: Use which error code? */
#endif /* _WIN32 */
}
+ /* MIPS has cache coherency issues, this is a no-op everywhere else
+ * Note: for any size >= on-chip cache size, entire on-chip cache is
+ * flushed.
+ */
+ CACHEFLUSH(env->me_map, txn->mt_next_pgno * env->me_psize, DCACHE);
+
for (i = keep; ++i <= pagecount; ) {
dp = dl[i].mptr;
/* This is a page we skipped above */
mdb_txn_commit(MDB_txn *txn)
{
int rc;
- unsigned int i;
+ unsigned int i, end_mode;
MDB_env *env;
- if (txn == NULL || txn->mt_env == NULL)
+ if (txn == NULL)
return EINVAL;
+ /* mdb_txn_end() mode for a commit which writes nothing */
+ end_mode = MDB_END_EMPTY_COMMIT|MDB_END_UPDATE|MDB_END_SLOT|MDB_END_FREE;
+
if (txn->mt_child) {
rc = mdb_txn_commit(txn->mt_child);
- txn->mt_child = NULL;
if (rc)
goto fail;
}
env = txn->mt_env;
if (F_ISSET(txn->mt_flags, MDB_TXN_RDONLY)) {
- mdb_dbis_update(txn, 1);
- txn->mt_numdbs = 2; /* so txn_abort() doesn't close any new handles */
- mdb_txn_abort(txn);
- return MDB_SUCCESS;
+ goto done;
}
- if (F_ISSET(txn->mt_flags, MDB_TXN_ERROR)) {
- DPUTS("error flag is set, can't commit");
+ if (txn->mt_flags & (MDB_TXN_FINISHED|MDB_TXN_ERROR)) {
+ DPUTS("txn has failed/finished, can't commit");
if (txn->mt_parent)
txn->mt_parent->mt_flags |= MDB_TXN_ERROR;
rc = MDB_BAD_TXN;
if (txn->mt_parent) {
MDB_txn *parent = txn->mt_parent;
+ MDB_page **lp;
MDB_ID2L dst, src;
MDB_IDL pspill;
unsigned x, y, len, ps_len;
/* Update parent's DB table. */
memcpy(parent->mt_dbs, txn->mt_dbs, txn->mt_numdbs * sizeof(MDB_db));
parent->mt_numdbs = txn->mt_numdbs;
- parent->mt_dbflags[0] = txn->mt_dbflags[0];
- parent->mt_dbflags[1] = txn->mt_dbflags[1];
- for (i=2; i<txn->mt_numdbs; i++) {
+ parent->mt_dbflags[FREE_DBI] = txn->mt_dbflags[FREE_DBI];
+ parent->mt_dbflags[MAIN_DBI] = txn->mt_dbflags[MAIN_DBI];
+ for (i=CORE_DBS; i<txn->mt_numdbs; i++) {
/* preserve parent's DB_NEW status */
x = parent->mt_dbflags[i] & DB_NEW;
parent->mt_dbflags[i] = txn->mt_dbflags[i] | x;
}
}
+ /* Append our loose page list to parent's */
+ for (lp = &parent->mt_loose_pgs; *lp; lp = &NEXT_LOOSE_PAGE(lp))
+ ;
+ *lp = txn->mt_loose_pgs;
+ parent->mt_loose_count += txn->mt_loose_count;
+
parent->mt_child = NULL;
mdb_midl_free(((MDB_ntxn *)txn)->mnt_pgstate.mf_pghead);
free(txn);
txn->mt_txnid, (void*)txn, (void*)env, txn->mt_dbs[MAIN_DBI].md_root));
/* Update DB root pointers */
- if (txn->mt_numdbs > 2) {
+ if (txn->mt_numdbs > CORE_DBS) {
MDB_cursor mc;
MDB_dbi i;
MDB_val data;
data.mv_size = sizeof(MDB_db);
mdb_cursor_init(&mc, txn, MAIN_DBI, NULL);
- for (i = 2; i < txn->mt_numdbs; i++) {
+ for (i = CORE_DBS; i < txn->mt_numdbs; i++) {
if (txn->mt_dbflags[i] & DB_DIRTY) {
+ if (TXN_DBI_CHANGED(txn, i)) {
+ rc = MDB_BAD_DBI;
+ goto fail;
+ }
data.mv_data = &txn->mt_dbs[i];
- rc = mdb_cursor_put(&mc, &txn->mt_dbxs[i].md_name, &data, 0);
+ rc = mdb_cursor_put(&mc, &txn->mt_dbxs[i].md_name, &data,
+ F_SUBDATA);
if (rc)
goto fail;
}
mdb_midl_free(env->me_pghead);
env->me_pghead = NULL;
- if (mdb_midl_shrink(&txn->mt_free_pgs))
- env->me_free_pgs = txn->mt_free_pgs;
+ mdb_midl_shrink(&txn->mt_free_pgs);
#if (MDB_DEBUG) > 2
mdb_audit(txn);
(rc = mdb_env_sync(env, 0)) ||
(rc = mdb_env_write_meta(txn)))
goto fail;
+ end_mode = MDB_END_COMMITTED|MDB_END_UPDATE;
done:
- env->me_pglast = 0;
- env->me_txn = NULL;
- mdb_dbis_update(txn, 1);
-
- if (env->me_txns)
- UNLOCK_MUTEX_W(env);
- free(txn);
-
+ mdb_txn_end(txn, end_mode);
return MDB_SUCCESS;
fail:
* @param[out] meta address of where to store the meta information
* @return 0 on success, non-zero on failure.
*/
-static int
+static int ESECT
mdb_env_read_header(MDB_env *env, MDB_meta *meta)
{
MDB_metabuf pbuf;
* Read both meta pages so we can use the latest one.
*/
- for (i=off=0; i<2; i++, off = meta->mm_psize) {
+ for (i=off=0; i<NUM_METAS; i++, off += meta->mm_psize) {
#ifdef _WIN32
DWORD len;
OVERLAPPED ov;
return 0;
}
+/** Fill in most of the zeroed #MDB_meta for an empty database environment */
+static void ESECT
+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 = NUM_METAS-1;
+ meta->mm_flags = env->me_flags & 0xffff;
+ meta->mm_flags |= MDB_INTEGERKEY; /* this is mm_dbs[FREE_DBI].md_flags */
+ meta->mm_dbs[FREE_DBI].md_root = P_INVALID;
+ meta->mm_dbs[MAIN_DBI].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
+ * @param[in] meta the #MDB_meta to write
* @return 0 on success, non-zero on failure.
*/
-static int
+static int ESECT
mdb_env_init_meta(MDB_env *env, MDB_meta *meta)
{
MDB_page *p, *q;
int len;
#define DO_PWRITE(rc, fd, ptr, size, len, pos) do { \
len = pwrite(fd, ptr, size, pos); \
- rc = (len >= 0); } while(0)
+ if (len == -1 && ErrCode() == EINTR) continue; \
+ rc = (len >= 0); break; } while(1)
#endif
DPUTS("writing new meta page");
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;
+ p = calloc(NUM_METAS, psize);
+ if (!p)
+ return ENOMEM;
- p = calloc(2, psize);
p->mp_pgno = 0;
p->mp_flags = P_META;
*(MDB_meta *)METADATA(p) = *meta;
q->mp_flags = P_META;
*(MDB_meta *)METADATA(q) = *meta;
- DO_PWRITE(rc, env->me_fd, p, psize * 2, len, 0);
+ DO_PWRITE(rc, env->me_fd, p, psize * NUM_METAS, len, 0);
if (!rc)
rc = ErrCode();
- else if ((unsigned) len == psize * 2)
+ else if ((unsigned) len == psize * NUM_METAS)
rc = MDB_SUCCESS;
else
rc = ENOSPC;
{
MDB_env *env;
MDB_meta meta, metab, *mp;
+ unsigned flags;
+ size_t mapsize;
off_t off;
int rc, len, toggle;
char *ptr;
toggle, txn->mt_dbs[MAIN_DBI].md_root));
env = txn->mt_env;
+ flags = env->me_flags;
mp = env->me_metas[toggle];
+ mapsize = env->me_metas[toggle ^ 1]->mm_mapsize;
+ /* Persist any increases of mapsize config */
+ if (mapsize < env->me_mapsize)
+ mapsize = env->me_mapsize;
- if (env->me_flags & MDB_WRITEMAP) {
- /* Persist any increases of mapsize config */
- if (env->me_mapsize > mp->mm_mapsize)
- mp->mm_mapsize = env->me_mapsize;
- mp->mm_dbs[0] = txn->mt_dbs[0];
- mp->mm_dbs[1] = txn->mt_dbs[1];
+ if (flags & MDB_WRITEMAP) {
+ mp->mm_mapsize = mapsize;
+ mp->mm_dbs[FREE_DBI] = txn->mt_dbs[FREE_DBI];
+ mp->mm_dbs[MAIN_DBI] = txn->mt_dbs[MAIN_DBI];
mp->mm_last_pg = txn->mt_next_pgno - 1;
+#if (__GNUC__ * 100 + __GNUC_MINOR__ >= 404) && /* TODO: portability */ \
+ !(defined(__i386__) || defined(__x86_64__))
+ /* LY: issue a memory barrier, if not x86. ITS#7969 */
+ __sync_synchronize();
+#endif
mp->mm_txnid = txn->mt_txnid;
- if (!(env->me_flags & (MDB_NOMETASYNC|MDB_NOSYNC))) {
+ if (!(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 = (char *)mp - PAGEHDRSZ;
#ifndef _WIN32 /* POSIX msync() requires ptr = start of OS page */
- if (meta_size < env->me_os_psize)
- meta_size += meta_size;
- else
+ r2 = (ptr - env->me_map) & (env->me_os_psize - 1);
+ ptr -= r2;
+ meta_size += r2;
#endif
- ptr += meta_size;
- }
if (MDB_MSYNC(ptr, meta_size, rc)) {
rc = ErrCode();
goto fail;
}
goto done;
}
- metab.mm_txnid = env->me_metas[toggle]->mm_txnid;
- metab.mm_last_pg = env->me_metas[toggle]->mm_last_pg;
-
- ptr = (char *)&meta;
- if (env->me_mapsize > mp->mm_mapsize) {
- /* Persist any increases of mapsize config */
- meta.mm_mapsize = env->me_mapsize;
- off = offsetof(MDB_meta, mm_mapsize);
- } else {
- off = offsetof(MDB_meta, mm_dbs[0].md_depth);
- }
- len = sizeof(MDB_meta) - off;
+ metab.mm_txnid = mp->mm_txnid;
+ metab.mm_last_pg = mp->mm_last_pg;
- ptr += off;
- meta.mm_dbs[0] = txn->mt_dbs[0];
- meta.mm_dbs[1] = txn->mt_dbs[1];
+ meta.mm_mapsize = mapsize;
+ meta.mm_dbs[FREE_DBI] = txn->mt_dbs[FREE_DBI];
+ meta.mm_dbs[MAIN_DBI] = txn->mt_dbs[MAIN_DBI];
meta.mm_last_pg = txn->mt_next_pgno - 1;
meta.mm_txnid = txn->mt_txnid;
- if (toggle)
- off += env->me_psize;
- off += PAGEHDRSZ;
+ off = offsetof(MDB_meta, mm_mapsize);
+ ptr = (char *)&meta + off;
+ len = sizeof(MDB_meta) - off;
+ off += (char *)mp - env->me_map;
/* Write to the SYNC fd */
- mfd = env->me_flags & (MDB_NOSYNC|MDB_NOMETASYNC) ?
- env->me_fd : env->me_mfd;
+ mfd = (flags & (MDB_NOSYNC|MDB_NOMETASYNC)) ? env->me_fd : env->me_mfd;
#ifdef _WIN32
{
memset(&ov, 0, sizeof(ov));
rc = -1;
}
#else
+retry_write:
rc = pwrite(mfd, ptr, len, off);
#endif
if (rc != len) {
rc = rc < 0 ? ErrCode() : EIO;
+#ifndef _WIN32
+ if (rc == EINTR)
+ goto retry_write;
+#endif
DPUTS("write failed, disk error?");
/* On a failure, the pagecache still contains the new data.
* Write some old data back, to prevent it from being used.
env->me_flags |= MDB_FATAL_ERROR;
return rc;
}
+ /* MIPS has cache coherency issues, this is a no-op everywhere else */
+ CACHEFLUSH(env->me_map + off, len, DCACHE);
done:
/* Memory ordering issues are irrelevant; since the entire writer
* is wrapped by wmutex, all of these changes will become visible
/** Check both meta pages to see which one is newer.
* @param[in] env the environment handle
- * @return meta toggle (0 or 1).
+ * @return newest #MDB_meta.
*/
-static int
+static MDB_meta *
mdb_env_pick_meta(const MDB_env *env)
{
- return (env->me_metas[0]->mm_txnid < env->me_metas[1]->mm_txnid);
+ MDB_meta *const *metas = env->me_metas;
+ return metas[ metas[0]->mm_txnid < metas[1]->mm_txnid ];
}
-int
+int ESECT
mdb_env_create(MDB_env **env)
{
MDB_env *e;
return ENOMEM;
e->me_maxreaders = DEFAULT_READERS;
- e->me_maxdbs = e->me_numdbs = 2;
+ e->me_maxdbs = e->me_numdbs = CORE_DBS;
e->me_fd = INVALID_HANDLE_VALUE;
e->me_lfd = INVALID_HANDLE_VALUE;
e->me_mfd = INVALID_HANDLE_VALUE;
return MDB_SUCCESS;
}
-static int
-mdb_env_map(MDB_env *env, void *addr, int newsize)
+static int ESECT
+mdb_env_map(MDB_env *env, void *addr)
{
MDB_page *p;
unsigned int flags = env->me_flags;
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;
- /* Windows won't create mappings for zero length files.
- * Just allocate the maxsize right now.
- */
- if (newsize) {
+ if (flags & MDB_RDONLY) {
+ /* Don't set explicit map size, use whatever exists */
+ 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.
+ * and won't map more than the file size.
+ * Just set the maxsize right now.
+ */
if (SetFilePointer(env->me_fd, sizelo, &sizehi, 0) != (DWORD)sizelo
|| !SetEndOfFile(env->me_fd)
|| SetFilePointer(env->me_fd, 0, NULL, 0) != 0)
return ErrCode();
}
+
mh = CreateFileMapping(env->me_fd, NULL, flags & MDB_WRITEMAP ?
PAGE_READWRITE : PAGE_READONLY,
sizehi, sizelo, NULL);
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)
return MDB_SUCCESS;
}
-int
+int ESECT
mdb_env_set_mapsize(MDB_env *env, size_t size)
{
/* If env is already open, caller is responsible for making
*/
if (env->me_map) {
int rc;
+ MDB_meta *meta;
void *old;
if (env->me_txn)
return EINVAL;
+ meta = mdb_env_pick_meta(env);
if (!size)
- size = env->me_metas[mdb_env_pick_meta(env)]->mm_mapsize;
- else if (size < env->me_mapsize) {
- /* If the configured size is smaller, make sure it's
- * still big enough. Silently round up to minimum if not.
- */
- size_t minsize = (env->me_metas[mdb_env_pick_meta(env)]->mm_last_pg + 1) * env->me_psize;
+ size = meta->mm_mapsize;
+ {
+ /* Silently round up to minimum if the size is too small */
+ size_t minsize = (meta->mm_last_pg + 1) * env->me_psize;
if (size < minsize)
size = minsize;
}
munmap(env->me_map, env->me_mapsize);
env->me_mapsize = size;
old = (env->me_flags & MDB_FIXEDMAP) ? env->me_map : NULL;
- rc = mdb_env_map(env, old, 1);
+ rc = mdb_env_map(env, old);
if (rc)
return rc;
}
return MDB_SUCCESS;
}
-int
+int ESECT
mdb_env_set_maxdbs(MDB_env *env, MDB_dbi dbs)
{
if (env->me_map)
return EINVAL;
- env->me_maxdbs = dbs + 2; /* Named databases + main and free DB */
+ env->me_maxdbs = dbs + CORE_DBS;
return MDB_SUCCESS;
}
-int
+int ESECT
mdb_env_set_maxreaders(MDB_env *env, unsigned int readers)
{
if (env->me_map || readers < 1)
return MDB_SUCCESS;
}
-int
+int ESECT
mdb_env_get_maxreaders(MDB_env *env, unsigned int *readers)
{
if (!env || !readers)
return MDB_SUCCESS;
}
-/** Further setup required for opening an MDB environment
- */
-static int
-mdb_env_open2(MDB_env *env)
+static int ESECT
+mdb_fsize(HANDLE fd, size_t *size)
{
- unsigned int flags = env->me_flags;
- int i, newenv = 0, rc;
- MDB_meta meta;
+#ifdef _WIN32
+ LARGE_INTEGER fsize;
+
+ if (!GetFileSizeEx(fd, &fsize))
+ return ErrCode();
+
+ *size = fsize.QuadPart;
+#else
+ struct stat st;
+
+ if (fstat(fd, &st))
+ return ErrCode();
+
+ *size = st.st_size;
+#endif
+ return MDB_SUCCESS;
+}
+
+#ifdef BROKEN_FDATASYNC
+#include <sys/utsname.h>
+#include <sys/vfs.h>
+#endif
+
+/** Further setup required for opening an LMDB environment
+ */
+static int ESECT
+mdb_env_open2(MDB_env *env)
+{
+ unsigned int flags = env->me_flags;
+ int i, newenv = 0, rc;
+ MDB_meta meta;
#ifdef _WIN32
/* See if we should use QueryLimited */
env->me_pidquery = PROCESS_QUERY_INFORMATION;
#endif /* _WIN32 */
- memset(&meta, 0, sizeof(meta));
+#ifdef BROKEN_FDATASYNC
+ /* ext3/ext4 fdatasync is broken on some older Linux kernels.
+ * https://lkml.org/lkml/2012/9/3/83
+ * Kernels after 3.6-rc6 are known good.
+ * https://lkml.org/lkml/2012/9/10/556
+ * See if the DB is on ext3/ext4, then check for new enough kernel
+ * Kernels 2.6.32.60, 2.6.34.15, 3.2.30, and 3.5.4 are also known
+ * to be patched.
+ */
+ {
+ struct statfs st;
+ fstatfs(env->me_fd, &st);
+ while (st.f_type == 0xEF53) {
+ struct utsname uts;
+ int i;
+ uname(&uts);
+ if (uts.release[0] < '3') {
+ if (!strncmp(uts.release, "2.6.32.", 7)) {
+ i = atoi(uts.release+7);
+ if (i >= 60)
+ break; /* 2.6.32.60 and newer is OK */
+ } else if (!strncmp(uts.release, "2.6.34.", 7)) {
+ i = atoi(uts.release+7);
+ if (i >= 15)
+ break; /* 2.6.34.15 and newer is OK */
+ }
+ } else if (uts.release[0] == '3') {
+ i = atoi(uts.release+2);
+ if (i > 5)
+ break; /* 3.6 and newer is OK */
+ if (i == 5) {
+ i = atoi(uts.release+4);
+ if (i >= 4)
+ break; /* 3.5.4 and newer is OK */
+ } else if (i == 2) {
+ i = atoi(uts.release+4);
+ if (i >= 30)
+ break; /* 3.2.30 and newer is OK */
+ }
+ } else { /* 4.x and newer is OK */
+ break;
+ }
+ env->me_flags |= MDB_FSYNCONLY;
+ break;
+ }
+ }
+#endif
if ((i = mdb_env_read_header(env, &meta)) != 0) {
if (i != ENOENT)
env->me_psize = env->me_os_psize;
if (env->me_psize > MAX_PAGESIZE)
env->me_psize = MAX_PAGESIZE;
+ memset(&meta, 0, sizeof(meta));
+ mdb_env_init_meta0(env, &meta);
+ meta.mm_mapsize = DEFAULT_MAPSIZE;
} else {
env->me_psize = meta.mm_psize;
}
/* Was a mapsize configured? */
if (!env->me_mapsize) {
- /* If this is a new environment, take the default,
- * else use the size recorded in the existing env.
- */
- env->me_mapsize = newenv ? DEFAULT_MAPSIZE : meta.mm_mapsize;
- } else if (env->me_mapsize < meta.mm_mapsize) {
- /* If the configured size is smaller, make sure it's
- * still big enough. Silently round up to minimum if not.
+ env->me_mapsize = meta.mm_mapsize;
+ }
+ {
+ /* Make sure mapsize >= committed data size. Even when using
+ * mm_mapsize, which could be broken in old files (ITS#7789).
*/
size_t minsize = (meta.mm_last_pg + 1) * meta.mm_psize;
if (env->me_mapsize < minsize)
env->me_mapsize = minsize;
}
+ meta.mm_mapsize = env->me_mapsize;
+
+ if (newenv && !(flags & MDB_FIXEDMAP)) {
+ /* mdb_env_map() may grow the datafile. Write the metapages
+ * first, so the file will be valid if initialization fails.
+ * Except with FIXEDMAP, since we do not yet know mm_address.
+ * We could fill in mm_address later, but then a different
+ * program might end up doing that - one with a memory layout
+ * and map address which does not suit the main program.
+ */
+ rc = mdb_env_init_meta(env, &meta);
+ if (rc)
+ return rc;
+ newenv = 0;
+ }
- rc = mdb_env_map(env, meta.mm_address, newenv || env->me_mapsize != meta.mm_mapsize);
+ rc = mdb_env_map(env, (flags & MDB_FIXEDMAP) ? meta.mm_address : NULL);
if (rc)
return rc;
#if MDB_DEBUG
{
- int toggle = mdb_env_pick_meta(env);
- MDB_db *db = &env->me_metas[toggle]->mm_dbs[MAIN_DBI];
+ MDB_meta *meta = mdb_env_pick_meta(env);
+ MDB_db *db = &meta->mm_dbs[MAIN_DBI];
DPRINTF(("opened database version %u, pagesize %u",
- env->me_metas[0]->mm_version, env->me_psize));
- DPRINTF(("using meta page %d", toggle));
+ meta->mm_version, env->me_psize));
+ DPRINTF(("using meta page %d", (int) (meta->mm_txnid & 1)));
DPRINTF(("depth: %u", db->md_depth));
DPRINTF(("entries: %"Z"u", db->md_entries));
DPRINTF(("branch pages: %"Z"u", db->md_branch_pages));
case DLL_THREAD_DETACH:
for (i=0; i<mdb_tls_nkeys; i++) {
MDB_reader *r = pthread_getspecific(mdb_tls_keys[i]);
- mdb_env_reader_dest(r);
+ if (r) {
+ mdb_env_reader_dest(r);
+ }
}
break;
case DLL_PROCESS_DETACH: break;
extern const PIMAGE_TLS_CALLBACK mdb_tls_cbp;
const PIMAGE_TLS_CALLBACK mdb_tls_cbp = mdb_tls_callback;
#pragma const_seg()
-#else /* WIN32 */
+#else /* _WIN32 */
#pragma comment(linker, "/INCLUDE:__tls_used")
#pragma comment(linker, "/INCLUDE:_mdb_tls_cbp")
#pragma data_seg(".CRT$XLB")
#endif
/** Downgrade the exclusive lock on the region back to shared */
-static int
+static int ESECT
mdb_env_share_locks(MDB_env *env, int *excl)
{
- int rc = 0, toggle = mdb_env_pick_meta(env);
+ int rc = 0;
+ MDB_meta *meta = mdb_env_pick_meta(env);
- env->me_txns->mti_txnid = env->me_metas[toggle]->mm_txnid;
+ env->me_txns->mti_txnid = meta->mm_txnid;
#ifdef _WIN32
{
return rc;
}
-/** Try to get exlusive lock, otherwise shared.
+/** Try to get exclusive lock, otherwise shared.
* Maintain *excl = -1: no/unknown lock, 0: shared, 1: exclusive.
*/
-static int
+static int ESECT
mdb_env_excl_lock(MDB_env *env, int *excl)
{
int rc = 0;
if (!rc) {
*excl = 1;
} else
-# ifdef MDB_USE_POSIX_SEM
- if (*excl < 0) /* always true when !MDB_USE_POSIX_SEM */
+# ifndef MDB_USE_POSIX_MUTEX
+ if (*excl < 0) /* always true when MDB_USE_POSIX_MUTEX */
# endif
{
lock_info.l_type = F_RDLCK;
*/
static const char mdb_a85[]= "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz!#$%&()*+-;<=>?@^_`{|}~";
-static void
+static void ESECT
mdb_pack85(unsigned long l, char *out)
{
int i;
}
}
-static void
+static void ESECT
mdb_hash_enc(MDB_val *val, char *encbuf)
{
mdb_hash_t h = mdb_hash_val(val, MDB_HASH_INIT);
#endif
/** Open and/or initialize the lock region for the environment.
- * @param[in] env The MDB environment.
+ * @param[in] env The LMDB environment.
* @param[in] lpath The pathname of the file used for the lock region.
* @param[in] mode The Unix permissions for the file, if we create it.
- * @param[out] excl Resulting file lock type: -1 none, 0 shared, 1 exclusive
* @param[in,out] excl In -1, out lock type: -1 none, 0 shared, 1 exclusive
* @return 0 on success, non-zero on failure.
*/
-static int
+static int ESECT
mdb_env_setup_locks(MDB_env *env, char *lpath, int mode, int *excl)
{
#ifdef _WIN32
env->me_wmutex = sem_open(env->me_txns->mti_wmname,
O_CREAT|O_EXCL, mode, 1);
if (env->me_wmutex == SEM_FAILED) goto fail_errno;
-#else /* MDB_USE_POSIX_SEM */
+#else /* MDB_USE_POSIX_MUTEX: */
pthread_mutexattr_t mattr;
if ((rc = pthread_mutexattr_init(&mattr))
|| (rc = pthread_mutexattr_setpshared(&mattr, PTHREAD_PROCESS_SHARED))
- || (rc = pthread_mutex_init(&env->me_txns->mti_mutex, &mattr))
- || (rc = pthread_mutex_init(&env->me_txns->mti_wmutex, &mattr)))
+#ifdef MDB_ROBUST_SUPPORTED
+ || (rc = pthread_mutexattr_setrobust(&mattr, PTHREAD_MUTEX_ROBUST))
+#endif
+ || (rc = pthread_mutex_init(env->me_txns->mti_rmutex, &mattr))
+ || (rc = pthread_mutex_init(env->me_txns->mti_wmutex, &mattr)))
goto fail;
pthread_mutexattr_destroy(&mattr);
#endif /* _WIN32 || MDB_USE_POSIX_SEM */
* environment and re-opening it with the new flags.
*/
#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)
+#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
+int ESECT
mdb_env_open(MDB_env *env, const char *path, unsigned int flags, mdb_mode_t mode)
{
int oflags, rc, len, excl = -1;
env->me_path = strdup(path);
env->me_dbxs = calloc(env->me_maxdbs, sizeof(MDB_dbx));
env->me_dbflags = calloc(env->me_maxdbs, sizeof(uint16_t));
- if (!(env->me_dbxs && env->me_path && env->me_dbflags)) {
+ env->me_dbiseqs = calloc(env->me_maxdbs, sizeof(unsigned int));
+ if (!(env->me_dbxs && env->me_path && env->me_dbflags && env->me_dbiseqs)) {
rc = ENOMEM;
goto leave;
}
+ env->me_dbxs[FREE_DBI].md_cmp = mdb_cmp_long; /* aligned MDB_INTEGERKEY */
/* For RDONLY, get lockfile after we know datafile exists */
if (!(flags & (MDB_RDONLY|MDB_NOLOCK))) {
if (rc)
goto leave;
}
- if (!((flags & MDB_RDONLY) ||
- (env->me_pbuf = calloc(1, env->me_psize))))
- rc = ENOMEM;
+ if (!(flags & MDB_RDONLY)) {
+ MDB_txn *txn;
+ int tsize = sizeof(MDB_txn), size = tsize + env->me_maxdbs *
+ (sizeof(MDB_db)+sizeof(MDB_cursor *)+sizeof(unsigned int)+1);
+ if ((env->me_pbuf = calloc(1, env->me_psize)) &&
+ (txn = calloc(1, size)))
+ {
+ txn->mt_dbs = (MDB_db *)((char *)txn + tsize);
+ txn->mt_cursors = (MDB_cursor **)(txn->mt_dbs + env->me_maxdbs);
+ txn->mt_dbiseqs = (unsigned int *)(txn->mt_cursors + env->me_maxdbs);
+ txn->mt_dbflags = (unsigned char *)(txn->mt_dbiseqs + env->me_maxdbs);
+ txn->mt_env = env;
+ txn->mt_dbxs = env->me_dbxs;
+ txn->mt_flags = MDB_TXN_FINISHED;
+ env->me_txn0 = txn;
+ } else {
+ rc = ENOMEM;
+ }
+ }
}
leave:
}
/** Destroy resources from mdb_env_open(), clear our readers & DBIs */
-static void
+static void ESECT
mdb_env_close0(MDB_env *env, int excl)
{
int i;
return;
/* Doing this here since me_dbxs may not exist during mdb_env_close */
- for (i = env->me_maxdbs; --i > MAIN_DBI; )
- free(env->me_dbxs[i].md_name.mv_data);
+ if (env->me_dbxs) {
+ for (i = env->me_maxdbs; --i >= CORE_DBS; )
+ free(env->me_dbxs[i].md_name.mv_data);
+ free(env->me_dbxs);
+ }
free(env->me_pbuf);
+ free(env->me_dbiseqs);
free(env->me_dbflags);
- free(env->me_dbxs);
free(env->me_path);
free(env->me_dirty_list);
+ free(env->me_txn0);
mdb_midl_free(env->me_free_pgs);
if (env->me_flags & MDB_ENV_TXKEY) {
MDB_PID_T pid = env->me_pid;
/* Clearing readers is done in this function because
* me_txkey with its destructor must be disabled first.
+ *
+ * We skip the the reader mutex, so we touch only
+ * data owned by this process (me_close_readers and
+ * our readers), and clear each reader atomically.
*/
- for (i = env->me_numreaders; --i >= 0; )
+ for (i = env->me_close_readers; --i >= 0; )
if (env->me_txns->mti_readers[i].mr_pid == pid)
env->me_txns->mti_readers[i].mr_pid = 0;
#ifdef _WIN32
env->me_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);
-
- 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;
-
- 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;
- }
- }
-
-leave:
- mdb_txn_abort(txn);
- return rc;
-}
-
-int
-mdb_env_copy(MDB_env *env, const char *path)
-{
- 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
+void ESECT
mdb_env_close(MDB_env *env)
{
MDB_page *dp;
*(size_t *)a->mv_data > *(size_t *)b->mv_data;
}
-/** Compare two items pointing at aligned unsigned int's */
+/** Compare two items pointing at aligned unsigned int's.
+ *
+ * This is also set as #MDB_INTEGERDUP|#MDB_DUPFIXED's #MDB_dbx.%md_dcmp,
+ * but #mdb_cmp_clong() is called instead if the data type is size_t.
+ */
static int
mdb_cmp_int(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
}
mdb_cursor_pop(MDB_cursor *mc)
{
if (mc->mc_snum) {
-#if MDB_DEBUG
- MDB_page *top = mc->mc_pg[mc->mc_top];
-#endif
+ DPRINTF(("popping page %"Z"u off db %d cursor %p",
+ mc->mc_pg[mc->mc_top]->mp_pgno, DDBI(mc), (void *) mc));
+
mc->mc_snum--;
if (mc->mc_snum)
mc->mc_top--;
-
- DPRINTF(("popped page %"Z"u off db %d cursor %p", top->mp_pgno,
- DDBI(mc), (void *) mc));
}
}
MDB_page *p = NULL;
int level;
- if (!((txn->mt_flags & MDB_TXN_RDONLY) | (env->me_flags & MDB_WRITEMAP))) {
+ if (! (txn->mt_flags & (MDB_TXN_RDONLY|MDB_TXN_WRITEMAP))) {
MDB_txn *tx2 = txn;
level = 1;
do {
/* Make sure the txn is still viable, then find the root from
* the txn's db table and set it as the root of the cursor's stack.
*/
- if (F_ISSET(mc->mc_txn->mt_flags, MDB_TXN_ERROR)) {
- DPUTS("transaction has failed, must abort");
+ if (mc->mc_txn->mt_flags & MDB_TXN_BLOCKED) {
+ DPUTS("transaction may not be used now");
return MDB_BAD_TXN;
} else {
/* Make sure we're using an up-to-date root */
if (*mc->mc_dbflag & DB_STALE) {
MDB_cursor mc2;
+ if (TXN_DBI_CHANGED(mc->mc_txn, mc->mc_dbi))
+ return MDB_BAD_DBI;
mdb_cursor_init(&mc2, mc->mc_txn, MAIN_DBI, NULL);
rc = mdb_page_search(&mc2, &mc->mc_dbx->md_name, 0);
if (rc)
&mc->mc_dbx->md_name, &exact);
if (!exact)
return MDB_NOTFOUND;
+ if ((leaf->mn_flags & (F_DUPDATA|F_SUBDATA)) != F_SUBDATA)
+ return MDB_INCOMPATIBLE; /* not a named DB */
rc = mdb_node_read(mc->mc_txn, leaf, &data);
if (rc)
return rc;
int exact = 0;
DKBUF;
- if (key == NULL || data == NULL)
- return EINVAL;
-
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 || !TXN_DBI_EXIST(txn, dbi, DB_USRVALID))
return EINVAL;
- if (txn->mt_flags & MDB_TXN_ERROR)
+ if (txn->mt_flags & MDB_TXN_BLOCKED)
return MDB_BAD_TXN;
mdb_cursor_init(&mc, txn, dbi, &mx);
if (op == MDB_PREV || op == MDB_PREV_DUP) {
rc = mdb_cursor_prev(&mc->mc_xcursor->mx_cursor, data, NULL, MDB_PREV);
if (op != MDB_PREV || rc != MDB_NOTFOUND) {
- if (rc == MDB_SUCCESS)
+ if (rc == MDB_SUCCESS) {
MDB_GET_KEY(leaf, key);
+ mc->mc_flags &= ~C_EOF;
+ }
return rc;
}
- } else {
- mc->mc_xcursor->mx_cursor.mc_flags &= ~(C_INITIALIZED|C_EOF);
- if (op == MDB_PREV_DUP)
- return MDB_NOTFOUND;
}
+ } else {
+ mc->mc_xcursor->mx_cursor.mc_flags &= ~(C_INITIALIZED|C_EOF);
+ if (op == MDB_PREV_DUP)
+ return MDB_NOTFOUND;
}
}
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
if (leaf == NULL) {
DPUTS("===> inexact leaf not found, goto sibling");
- if ((rc = mdb_cursor_sibling(mc, 1)) != MDB_SUCCESS)
+ if ((rc = mdb_cursor_sibling(mc, 1)) != MDB_SUCCESS) {
+ mc->mc_flags |= C_EOF;
return rc; /* no entries matched */
+ }
mp = mc->mc_pg[mc->mc_top];
mdb_cassert(mc, IS_LEAF(mp));
leaf = NODEPTR(mp, 0);
mc->mc_flags &= ~C_EOF;
if (IS_LEAF2(mp)) {
- key->mv_size = mc->mc_db->md_pad;
- key->mv_data = LEAF2KEY(mp, mc->mc_ki[mc->mc_top], key->mv_size);
+ if (op == MDB_SET_RANGE || op == MDB_SET_KEY) {
+ key->mv_size = mc->mc_db->md_pad;
+ key->mv_data = LEAF2KEY(mp, mc->mc_ki[mc->mc_top], key->mv_size);
+ }
return MDB_SUCCESS;
}
return rc;
}
} else if (op == MDB_GET_BOTH || op == MDB_GET_BOTH_RANGE) {
- MDB_val d2;
- if ((rc = mdb_node_read(mc->mc_txn, leaf, &d2)) != MDB_SUCCESS)
+ MDB_val olddata;
+ MDB_cmp_func *dcmp;
+ if ((rc = mdb_node_read(mc->mc_txn, leaf, &olddata)) != MDB_SUCCESS)
return rc;
- rc = mc->mc_dbx->md_dcmp(data, &d2);
+ dcmp = mc->mc_dbx->md_dcmp;
+#if UINT_MAX < SIZE_MAX
+ if (dcmp == mdb_cmp_int && olddata.mv_size == sizeof(size_t))
+ dcmp = mdb_cmp_clong;
+#endif
+ rc = dcmp(data, &olddata);
if (rc) {
if (op == MDB_GET_BOTH || rc > 0)
return MDB_NOTFOUND;
rc = 0;
- *data = d2;
+ *data = olddata;
}
} else {
if (mc == NULL)
return EINVAL;
- if (mc->mc_txn->mt_flags & MDB_TXN_ERROR)
+ if (mc->mc_txn->mt_flags & MDB_TXN_BLOCKED)
return MDB_BAD_TXN;
switch (op) {
rc = MDB_INCOMPATIBLE;
break;
}
+ {
+ MDB_node *leaf = NODEPTR(mc->mc_pg[mc->mc_top], mc->mc_ki[mc->mc_top]);
+ if (!F_ISSET(leaf->mn_flags, F_DUPDATA)) {
+ MDB_GET_KEY(leaf, key);
+ rc = mdb_node_read(mc->mc_txn, leaf, data);
+ break;
+ }
+ }
if (!(mc->mc_xcursor->mx_cursor.mc_flags & C_INITIALIZED)) {
rc = EINVAL;
break;
{
int rc = MDB_SUCCESS;
- if (mc->mc_dbi > MAIN_DBI && !(*mc->mc_dbflag & DB_DIRTY)) {
+ if (mc->mc_dbi >= CORE_DBS && !(*mc->mc_dbflag & DB_DIRTY)) {
MDB_cursor mc2;
MDB_xcursor mcx;
+ if (TXN_DBI_CHANGED(mc->mc_txn, mc->mc_dbi))
+ return MDB_BAD_DBI;
mdb_cursor_init(&mc2, mc->mc_txn, MAIN_DBI, &mcx);
rc = mdb_page_search(&mc2, &mc->mc_dbx->md_name, MDB_PS_MODIFY);
if (rc)
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_page *fp, *mp;
+ MDB_page *fp, *mp, *sub_root = NULL;
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)
+ if (mc == NULL || key == NULL)
return EINVAL;
env = mc->mc_txn->mt_env;
nospill = flags & MDB_NOSPILL;
flags &= ~MDB_NOSPILL;
- if (mc->mc_txn->mt_flags & (MDB_TXN_RDONLY|MDB_TXN_ERROR))
+ if (mc->mc_txn->mt_flags & (MDB_TXN_RDONLY|MDB_TXN_BLOCKED))
return (mc->mc_txn->mt_flags & MDB_TXN_RDONLY) ? EACCES : MDB_BAD_TXN;
- if (flags != MDB_CURRENT) {
- if (key == NULL)
- return EINVAL;
- if (key->mv_size-1 >= ENV_MAXKEY(env))
- return MDB_BAD_VALSIZE;
- } else {
- /* Ignore key except in sub-cursor, where key holds the data */
- if (!(mc->mc_flags & C_SUB))
- key = NULL;
- }
+ if (key->mv_size-1 >= ENV_MAXKEY(env))
+ return MDB_BAD_VALSIZE;
#if SIZE_MAX > MAXDATASIZE
if (data->mv_size > ((mc->mc_db->md_flags & MDB_DUPSORT) ? ENV_MAXKEY(env) : MAXDATASIZE))
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) &&
LEAFSIZE(key, data) > env->me_nodemax)
{
- /* Too big for a node, insert in sub-DB */
+ /* Too big for a node, insert in sub-DB. Set up an empty
+ * "old sub-page" for prep_subDB to expand to a full page.
+ */
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;
+ fp->mp_lower = fp->mp_upper = (PAGEHDRSZ-PAGEBASE);
+ 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);
+fix_parent:
+ /* if overwriting slot 0 of leaf, need to
+ * update branch key if there is a parent page
+ */
+ if (mc->mc_top && !mc->mc_ki[mc->mc_top]) {
+ unsigned short top = mc->mc_top;
+ mc->mc_top--;
+ /* slot 0 is always an empty key, find real slot */
+ while (mc->mc_top && !mc->mc_ki[mc->mc_top])
+ mc->mc_top--;
+ if (mc->mc_ki[mc->mc_top])
+ rc2 = mdb_update_key(mc, key);
+ else
+ rc2 = MDB_SUCCESS;
+ mc->mc_top = top;
+ if (rc2)
+ return rc2;
}
return MDB_SUCCESS;
}
/* Was a single item before, must convert now */
if (!F_ISSET(leaf->mn_flags, F_DUPDATA)) {
+ MDB_cmp_func *dcmp;
/* Just overwrite the current item */
if (flags == MDB_CURRENT)
goto current;
-
+ dcmp = mc->mc_dbx->md_dcmp;
#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;
-#endif
+ if (dcmp == mdb_cmp_int && olddata.mv_size == sizeof(size_t))
+ dcmp = mdb_cmp_clong;
#endif
- /* if data matches, skip it */
- if (!mc->mc_dbx->md_dcmp(data, &olddata)) {
+ /* does data match? */
+ if (!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 */
/* 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_lower = (PAGEHDRSZ-PAGEBASE);
xdata.mv_size = PAGEHDRSZ + dkey.mv_size + data->mv_size;
if (mc->mc_db->md_flags & MDB_DUPFIXED) {
fp->mp_flags |= P_LEAF2;
xdata.mv_size += 2 * (sizeof(indx_t) + NODESIZE) +
(dkey.mv_size & 1) + (data->mv_size & 1);
}
- fp->mp_upper = xdata.mv_size;
- olddata.mv_size = fp->mp_upper; /* pretend olddata is fp */
+ fp->mp_upper = xdata.mv_size - PAGEBASE;
+ olddata.mv_size = xdata.mv_size; /* pretend olddata is fp */
} else if (leaf->mn_flags & F_SUBDATA) {
/* Data is on sub-DB, just store it */
flags |= F_DUPDATA|F_SUBDATA;
offset = env->me_psize - olddata.mv_size;
flags |= F_DUPDATA|F_SUBDATA;
dummy.md_root = mp->mp_pgno;
+ sub_root = mp;
}
if (mp != fp) {
mp->mp_flags = fp_flags | P_DIRTY;
if (fp_flags & P_LEAF2) {
memcpy(METADATA(mp), METADATA(fp), NUMKEYS(fp) * fp->mp_pad);
} else {
- memcpy((char *)mp + mp->mp_upper, (char *)fp + fp->mp_upper,
- olddata.mv_size - fp->mp_upper);
+ memcpy((char *)mp + mp->mp_upper + PAGEBASE, (char *)fp + fp->mp_upper + PAGEBASE,
+ olddata.mv_size - fp->mp_upper - PAGEBASE);
for (i=0; i<NUMKEYS(fp); i++)
mp->mp_ptrs[i] = fp->mp_ptrs[i] + offset;
}
rdata = &xdata;
flags |= F_DUPDATA;
do_sub = 1;
- if (!insert)
+ if (!insert_key)
mdb_node_del(mc, 0);
goto new_sub;
}
current:
+ /* LMDB passes F_SUBDATA in 'flags' to write a DB record */
+ if ((leaf->mn_flags ^ flags) & F_SUBDATA)
+ return MDB_INCOMPATIBLE;
/* overflow page overwrites need special handling */
if (F_ISSET(leaf->mn_flags, F_BIGDATA)) {
MDB_page *omp;
return ENOMEM;
id2.mid = pg;
id2.mptr = np;
- rc = mdb_mid2l_insert(mc->mc_txn->mt_u.dirty_list, &id2);
- mdb_cassert(mc, rc == 0);
+ rc2 = mdb_mid2l_insert(mc->mc_txn->mt_u.dirty_list, &id2);
+ mdb_cassert(mc, rc2 == 0);
if (!(flags & MDB_RESERVE)) {
/* Copy end of page, adjusting alignment so
* compiler may copy words instead of bytes.
data->mv_data = METADATA(omp);
else
memcpy(METADATA(omp), data->mv_data, data->mv_size);
- goto done;
+ return MDB_SUCCESS;
}
}
if ((rc2 = mdb_ovpage_free(mc, omp)) != MDB_SUCCESS)
data->mv_data = olddata.mv_data;
else if (!(mc->mc_flags & C_SUB))
memcpy(olddata.mv_data, data->mv_data, data->mv_size);
- else
+ else {
memcpy(NODEKEY(leaf), key->mv_data, key->mv_size);
- goto done;
+ goto fix_parent;
+ }
+ return MDB_SUCCESS;
}
mdb_node_del(mc, 0);
- mc->mc_db->md_entries--;
}
rdata = data;
nsize = IS_LEAF2(mc->mc_pg[mc->mc_top]) ? key->mv_size : mdb_leaf_size(env, key, rdata);
if (SIZELEFT(mc->mc_pg[mc->mc_top]) < nsize) {
if (( flags & (F_DUPDATA|F_SUBDATA)) == F_DUPDATA )
- nflags &= ~MDB_APPEND;
- if (!insert)
+ nflags &= ~MDB_APPEND; /* sub-page may need room to grow */
+ if (!insert_key)
nflags |= MDB_SPLIT_REPLACE;
rc = mdb_page_split(mc, key, rdata, P_INVALID, nflags);
} else {
/* There is room already in this leaf page. */
rc = mdb_node_add(mc, mc->mc_ki[mc->mc_top], key, rdata, 0, nflags);
- if (rc == 0 && !do_sub && insert) {
+ if (rc == 0 && insert_key) {
/* Adjust other cursors pointing to mp */
MDB_cursor *m2, *m3;
MDB_dbi dbi = mc->mc_dbi;
}
}
- if (rc != MDB_SUCCESS)
- mc->mc_txn->mt_flags |= MDB_TXN_ERROR;
- else {
+ if (rc == MDB_SUCCESS) {
/* Now store the actual data in the child DB. Note that we're
* storing the user data in the keys field, so there are strict
* size limits on dupdata. The actual data fields of the child
* DB are all zero size.
*/
if (do_sub) {
- int xflags;
+ int xflags, new_dupdata;
+ size_t ecount;
put_sub:
xdata.mv_size = 0;
xdata.mv_data = "";
xflags = (flags & MDB_NODUPDATA) ?
MDB_NOOVERWRITE|MDB_NOSPILL : MDB_NOSPILL;
}
+ if (sub_root)
+ mc->mc_xcursor->mx_cursor.mc_pg[0] = sub_root;
+ new_dupdata = (int)dkey.mv_size;
/* converted, write the original data first */
if (dkey.mv_size) {
rc = mdb_cursor_put(&mc->mc_xcursor->mx_cursor, &dkey, &xdata, xflags);
if (rc)
- return rc;
- {
- /* Adjust other cursors pointing to mp */
- MDB_cursor *m2;
- unsigned i = mc->mc_top;
- MDB_page *mp = mc->mc_pg[i];
-
- 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_flags & C_INITIALIZED)) continue;
- if (m2->mc_pg[i] == mp && m2->mc_ki[i] == mc->mc_ki[i]) {
- mdb_xcursor_init1(m2, leaf);
+ goto bad_sub;
+ /* we've done our job */
+ dkey.mv_size = 0;
+ }
+ if (!(leaf->mn_flags & F_SUBDATA) || sub_root) {
+ /* Adjust other cursors pointing to mp */
+ MDB_cursor *m2;
+ MDB_xcursor *mx = mc->mc_xcursor;
+ unsigned i = mc->mc_top;
+ MDB_page *mp = mc->mc_pg[i];
+
+ 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_flags & C_INITIALIZED)) continue;
+ if (m2->mc_pg[i] == mp) {
+ if (m2->mc_ki[i] == mc->mc_ki[i]) {
+ mdb_xcursor_init2(m2, mx, new_dupdata);
+ } else if (!insert_key) {
+ MDB_node *n2 = NODEPTR(mp, m2->mc_ki[i]);
+ if (!(n2->mn_flags & F_SUBDATA))
+ m2->mc_xcursor->mx_cursor.mc_pg[0] = NODEDATA(n2);
}
}
}
- /* we've done our job */
- dkey.mv_size = 0;
}
+ ecount = mc->mc_xcursor->mx_db.md_entries;
if (flags & MDB_APPENDDUP)
xflags |= MDB_APPEND;
rc = mdb_cursor_put(&mc->mc_xcursor->mx_cursor, data, &xdata, xflags);
void *db = NODEDATA(leaf);
memcpy(db, &mc->mc_xcursor->mx_db, sizeof(MDB_db));
}
+ insert_data = mc->mc_xcursor->mx_db.md_entries - ecount;
}
- /* sub-writes might have failed so check rc again.
- * Don't increment count if we just replaced an existing item.
- */
- if (!rc && !(flags & MDB_CURRENT))
+ /* Increment count unless we just replaced an existing item. */
+ if (insert_data)
mc->mc_db->md_entries++;
+ if (insert_key) {
+ /* Invalidate txn if we created an empty sub-DB */
+ if (rc)
+ goto bad_sub;
+ /* If we succeeded and the key didn't exist before,
+ * make sure the cursor is marked valid.
+ */
+ mc->mc_flags |= C_INITIALIZED;
+ }
if (flags & MDB_MULTIPLE) {
if (!rc) {
-next_mult:
mcount++;
/* let caller know how many succeeded, if any */
data[1].mv_size = mcount;
if (mcount < dcount) {
data[0].mv_data = (char *)data[0].mv_data + data[0].mv_size;
+ 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;
}
MDB_page *mp;
int rc;
- if (mc->mc_txn->mt_flags & (MDB_TXN_RDONLY|MDB_TXN_ERROR))
+ if (mc->mc_txn->mt_flags & (MDB_TXN_RDONLY|MDB_TXN_BLOCKED))
return (mc->mc_txn->mt_flags & MDB_TXN_RDONLY) ? EACCES : MDB_BAD_TXN;
if (!(mc->mc_flags & C_INITIALIZED))
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) {
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 */
+ /* fix other sub-DB cursors pointed at fake pages on this 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] == mp &&
- m2->mc_ki[mc->mc_top] == mc->mc_ki[mc->mc_top])
- m2->mc_xcursor->mx_cursor.mc_pg[0] = NODEDATA(leaf);
+ if (!(m2->mc_flags & C_INITIALIZED)) continue;
+ if (m2->mc_pg[mc->mc_top] == mp) {
+ if (m2->mc_ki[mc->mc_top] == mc->mc_ki[mc->mc_top]) {
+ m2->mc_xcursor->mx_cursor.mc_pg[0] = NODEDATA(leaf);
+ } else {
+ MDB_node *n2 = NODEPTR(mp, m2->mc_ki[mc->mc_top]);
+ if (!(n2->mn_flags & F_SUBDATA))
+ m2->mc_xcursor->mx_cursor.mc_pg[0] = NODEDATA(n2);
+ }
+ }
}
}
mc->mc_db->md_entries--;
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;
}
}
+ /* LMDB passes F_SUBDATA in 'flags' to delete a DB record */
+ else if ((leaf->mn_flags ^ flags) & F_SUBDATA) {
+ rc = MDB_INCOMPATIBLE;
+ goto fail;
+ }
+
+ /* 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);
- return mdb_cursor_del0(mc, leaf);
+fail:
+ mc->mc_txn->mt_flags |= MDB_TXN_ERROR;
+ return rc;
}
/** Allocate and initialize new pages for a database.
DPRINTF(("allocated new mpage %"Z"u, page size %u",
np->mp_pgno, mc->mc_txn->mt_env->me_psize));
np->mp_flags = flags | P_DIRTY;
- np->mp_lower = PAGEHDRSZ;
- np->mp_upper = mc->mc_txn->mt_env->me_psize;
+ np->mp_lower = (PAGEHDRSZ-PAGEBASE);
+ np->mp_upper = mc->mc_txn->mt_env->me_psize - PAGEBASE;
if (IS_BRANCH(np))
mc->mc_db->md_branch_pages++;
MDB_node *node;
MDB_page *mp = mc->mc_pg[mc->mc_top];
MDB_page *ofp = NULL; /* overflow page */
+ void *ndata;
DKBUF;
mdb_cassert(mc, mp->mp_upper >= mp->mp_lower);
if (key != NULL)
node_size += key->mv_size;
if (IS_LEAF(mp)) {
- mdb_cassert(mc, data);
+ mdb_cassert(mc, key && data);
if (F_ISSET(flags, F_BIGDATA)) {
/* Data already on overflow page. */
node_size += sizeof(pgno_t);
memcpy(NODEKEY(node), key->mv_data, key->mv_size);
if (IS_LEAF(mp)) {
- mdb_cassert(mc, key);
+ ndata = NODEDATA(node);
if (ofp == NULL) {
if (F_ISSET(flags, F_BIGDATA))
- memcpy(node->mn_data + key->mv_size, data->mv_data,
- sizeof(pgno_t));
+ memcpy(ndata, data->mv_data, sizeof(pgno_t));
else if (F_ISSET(flags, MDB_RESERVE))
- data->mv_data = node->mn_data + key->mv_size;
+ data->mv_data = ndata;
else
- memcpy(node->mn_data + key->mv_size, data->mv_data,
- data->mv_size);
+ memcpy(ndata, data->mv_data, data->mv_size);
} else {
- memcpy(node->mn_data + key->mv_size, &ofp->mp_pgno,
- sizeof(pgno_t));
+ memcpy(ndata, &ofp->mp_pgno, sizeof(pgno_t));
+ ndata = METADATA(ofp);
if (F_ISSET(flags, MDB_RESERVE))
- data->mv_data = METADATA(ofp);
+ data->mv_data = ndata;
else
- memcpy(METADATA(ofp), data->mv_data, data->mv_size);
+ memcpy(ndata, data->mv_data, data->mv_size);
}
}
}
}
- base = (char *)mp + mp->mp_upper;
+ base = (char *)mp + mp->mp_upper + PAGEBASE;
memmove(base + sz, base, ptr - mp->mp_upper);
mp->mp_lower -= sizeof(indx_t);
MDB_node *node;
MDB_page *sp, *xp;
char *base;
- int nsize, delta;
- indx_t i, numkeys, ptr;
+ indx_t delta, nsize, len, ptr;
+ int i;
node = NODEPTR(mp, indx);
sp = (MDB_page *)NODEDATA(node);
delta = SIZELEFT(sp);
- xp = (MDB_page *)((char *)sp + delta);
+ nsize = NODEDSZ(node) - delta;
- /* shift subpage upward */
+ /* Prepare to shift upward, set len = length(subpage part to shift) */
if (IS_LEAF2(sp)) {
- nsize = NUMKEYS(sp) * sp->mp_pad;
+ len = nsize;
if (nsize & 1)
return; /* do not make the node uneven-sized */
- memmove(METADATA(xp), METADATA(sp), nsize);
} else {
- int i;
- numkeys = NUMKEYS(sp);
- for (i=numkeys-1; i>=0; i--)
+ xp = (MDB_page *)((char *)sp + delta); /* destination subpage */
+ for (i = NUMKEYS(sp); --i >= 0; )
xp->mp_ptrs[i] = sp->mp_ptrs[i] - delta;
+ len = PAGEHDRSZ;
}
- xp->mp_upper = sp->mp_lower;
- xp->mp_lower = sp->mp_lower;
- xp->mp_flags = sp->mp_flags;
- xp->mp_pad = sp->mp_pad;
- COPY_PGNO(xp->mp_pgno, mp->mp_pgno);
-
- nsize = NODEDSZ(node) - delta;
+ sp->mp_upper = sp->mp_lower;
+ COPY_PGNO(sp->mp_pgno, mp->mp_pgno);
SETDSZ(node, nsize);
- /* shift lower nodes upward */
+ /* Shift <lower nodes...initial part of subpage> upward */
+ base = (char *)mp + mp->mp_upper + PAGEBASE;
+ memmove(base + delta, base, (char *)sp + len - base);
+
ptr = mp->mp_ptrs[indx];
- numkeys = NUMKEYS(mp);
- for (i = 0; i < numkeys; i++) {
+ for (i = NUMKEYS(mp); --i >= 0; ) {
if (mp->mp_ptrs[i] <= ptr)
mp->mp_ptrs[i] += delta;
}
-
- base = (char *)mp + mp->mp_upper;
- memmove(base + delta, base, ptr - mp->mp_upper + NODESIZE + NODEKSZ(node));
mp->mp_upper += delta;
}
mx->mx_cursor.mc_flags = C_SUB;
} else {
MDB_page *fp = NODEDATA(node);
- mx->mx_db.md_pad = mc->mc_pg[mc->mc_top]->mp_pad;
+ mx->mx_db.md_pad = 0;
mx->mx_db.md_flags = 0;
mx->mx_db.md_depth = 1;
mx->mx_db.md_branch_pages = 0;
}
DPRINTF(("Sub-db -%u root page %"Z"u", mx->mx_cursor.mc_dbi,
mx->mx_db.md_root));
- mx->mx_dbflag = DB_VALID|DB_DIRTY; /* DB_DIRTY guides mdb_cursor_touch */
+ mx->mx_dbflag = DB_VALID|DB_USRVALID|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;
+ mx->mx_dbx.md_cmp = mdb_cmp_clong;
#endif
+}
+
+
+/** Fixup a sorted-dups cursor due to underlying update.
+ * Sets up some fields that depend on the data from the main cursor.
+ * Almost the same as init1, but skips initialization steps if the
+ * xcursor had already been used.
+ * @param[in] mc The main cursor whose sorted-dups cursor is to be fixed up.
+ * @param[in] src_mx The xcursor of an up-to-date cursor.
+ * @param[in] new_dupdata True if converting from a non-#F_DUPDATA item.
+ */
+static void
+mdb_xcursor_init2(MDB_cursor *mc, MDB_xcursor *src_mx, int new_dupdata)
+{
+ MDB_xcursor *mx = mc->mc_xcursor;
+
+ if (new_dupdata) {
+ mx->mx_cursor.mc_snum = 1;
+ mx->mx_cursor.mc_top = 0;
+ mx->mx_cursor.mc_flags |= C_INITIALIZED;
+ mx->mx_cursor.mc_ki[0] = 0;
+ mx->mx_dbflag = DB_VALID|DB_USRVALID|DB_DIRTY; /* DB_DIRTY guides mdb_cursor_touch */
+#if UINT_MAX < SIZE_MAX
+ mx->mx_dbx.md_cmp = src_mx->mx_dbx.md_cmp;
#endif
+ } else if (!(mx->mx_cursor.mc_flags & C_INITIALIZED)) {
+ return;
+ }
+ mx->mx_db = src_mx->mx_db;
+ mx->mx_cursor.mc_pg[0] = src_mx->mx_cursor.mc_pg[0];
+ DPRINTF(("Sub-db -%u root page %"Z"u", mx->mx_cursor.mc_dbi,
+ mx->mx_db.md_root));
}
/** Initialize a cursor for a given transaction and database. */
mc->mc_snum = 0;
mc->mc_top = 0;
mc->mc_pg[0] = 0;
+ mc->mc_ki[0] = 0;
mc->mc_flags = 0;
if (txn->mt_dbs[dbi].md_flags & MDB_DUPSORT) {
mdb_tassert(txn, mx != NULL);
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, DB_VALID))
return EINVAL;
- if (txn->mt_flags & MDB_TXN_ERROR)
+ if (txn->mt_flags & MDB_TXN_BLOCKED)
return MDB_BAD_TXN;
- /* Allow read access to the freelist */
- if (!dbi && !F_ISSET(txn->mt_flags, MDB_TXN_RDONLY))
+ if (dbi == FREE_DBI && !F_ISSET(txn->mt_flags, MDB_TXN_RDONLY))
return EINVAL;
if (txn->mt_dbs[dbi].md_flags & MDB_DUPSORT)
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, DB_VALID))
return EINVAL;
if ((mc->mc_flags & C_UNTRACK) || txn->mt_cursors)
return EINVAL;
+ if (txn->mt_flags & MDB_TXN_BLOCKED)
+ return MDB_BAD_TXN;
+
mdb_cursor_init(mc, txn, mc->mc_dbi, mc->mc_xcursor);
return MDB_SUCCESS;
}
if (mc->mc_xcursor == NULL)
return MDB_INCOMPATIBLE;
+ if (mc->mc_txn->mt_flags & MDB_TXN_BLOCKED)
+ return MDB_BAD_TXN;
+
+ if (!(mc->mc_flags & C_INITIALIZED))
+ return EINVAL;
+
+ if (!mc->mc_snum || (mc->mc_flags & C_EOF))
+ return MDB_NOTFOUND;
+
leaf = NODEPTR(mc->mc_pg[mc->mc_top], mc->mc_ki[mc->mc_top]);
if (!F_ISSET(leaf->mn_flags, F_DUPDATA)) {
*countp = 1;
mp->mp_ptrs[i] -= delta;
}
- base = (char *)mp + mp->mp_upper;
+ base = (char *)mp + mp->mp_upper + PAGEBASE;
len = ptr - mp->mp_upper + NODESIZE;
memmove(base - delta, base, len);
mp->mp_upper -= delta;
unsigned int snum = csrc->mc_snum;
MDB_node *s2;
/* must find the lowest key below src */
- mdb_page_search_lowest(csrc);
+ rc = mdb_page_search_lowest(csrc);
+ if (rc)
+ return rc;
if (IS_LEAF2(csrc->mc_pg[csrc->mc_top])) {
key.mv_size = csrc->mc_db->md_pad;
key.mv_data = LEAF2KEY(csrc->mc_pg[csrc->mc_top], 0, key.mv_size);
MDB_node *s2;
MDB_val bkey;
/* must find the lowest key below dst */
- mdb_page_search_lowest(cdst);
- if (IS_LEAF2(cdst->mc_pg[cdst->mc_top])) {
- bkey.mv_size = cdst->mc_db->md_pad;
- bkey.mv_data = LEAF2KEY(cdst->mc_pg[cdst->mc_top], 0, bkey.mv_size);
+ mdb_cursor_copy(cdst, &mn);
+ rc = mdb_page_search_lowest(&mn);
+ if (rc)
+ return rc;
+ if (IS_LEAF2(mn.mc_pg[mn.mc_top])) {
+ bkey.mv_size = mn.mc_db->md_pad;
+ bkey.mv_data = LEAF2KEY(mn.mc_pg[mn.mc_top], 0, bkey.mv_size);
} else {
- s2 = NODEPTR(cdst->mc_pg[cdst->mc_top], 0);
+ s2 = NODEPTR(mn.mc_pg[mn.mc_top], 0);
bkey.mv_size = NODEKSZ(s2);
bkey.mv_data = NODEKEY(s2);
}
- cdst->mc_snum = snum--;
- cdst->mc_top = snum;
- mdb_cursor_copy(cdst, &mn);
+ mn.mc_snum = snum--;
+ mn.mc_top = snum;
mn.mc_ki[snum] = 0;
rc = mdb_update_key(&mn, &bkey);
if (rc)
/* Adjust other cursors pointing to mp */
MDB_cursor *m2, *m3;
MDB_dbi dbi = csrc->mc_dbi;
- MDB_page *mp = csrc->mc_pg[csrc->mc_top];
+ MDB_page *mp;
+
+ mp = cdst->mc_pg[csrc->mc_top];
+ for (m2 = csrc->mc_txn->mt_cursors[dbi]; m2; m2=m2->mc_next) {
+ if (csrc->mc_flags & C_SUB)
+ m3 = &m2->mc_xcursor->mx_cursor;
+ else
+ m3 = m2;
+ if (m3 == cdst) continue;
+ if (m3->mc_pg[csrc->mc_top] == mp && m3->mc_ki[csrc->mc_top] >=
+ cdst->mc_ki[csrc->mc_top]) {
+ m3->mc_ki[csrc->mc_top]++;
+ }
+ }
+ mp = csrc->mc_pg[csrc->mc_top];
for (m2 = csrc->mc_txn->mt_cursors[dbi]; m2; m2=m2->mc_next) {
if (csrc->mc_flags & C_SUB)
m3 = &m2->mc_xcursor->mx_cursor;
cdst->mc_ki[cdst->mc_top] = 0;
rc = mdb_update_key(cdst, &nullkey);
cdst->mc_ki[cdst->mc_top] = ix;
- mdb_cassert(csrc, rc == MDB_SUCCESS);
+ mdb_cassert(cdst, rc == MDB_SUCCESS);
}
}
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));
mdb_cassert(csrc, csrc->mc_snum > 1); /* can't merge root page */
mdb_cassert(csrc, cdst->mc_snum > 1);
/* Move all nodes from src to dst.
*/
- j = nkeys = NUMKEYS(cdst->mc_pg[cdst->mc_top]);
- if (IS_LEAF2(csrc->mc_pg[csrc->mc_top])) {
+ j = nkeys = NUMKEYS(pdst);
+ if (IS_LEAF2(psrc)) {
key.mv_size = csrc->mc_db->md_pad;
- key.mv_data = METADATA(csrc->mc_pg[csrc->mc_top]);
- for (i = 0; i < NUMKEYS(csrc->mc_pg[csrc->mc_top]); i++, j++) {
+ key.mv_data = METADATA(psrc);
+ for (i = 0; i < NUMKEYS(psrc); i++, j++) {
rc = mdb_node_add(cdst, j, &key, NULL, 0, 0);
if (rc != MDB_SUCCESS)
return rc;
key.mv_data = (char *)key.mv_data + key.mv_size;
}
} else {
- for (i = 0; i < NUMKEYS(csrc->mc_pg[csrc->mc_top]); i++, j++) {
- srcnode = NODEPTR(csrc->mc_pg[csrc->mc_top], i);
- if (i == 0 && IS_BRANCH(csrc->mc_pg[csrc->mc_top])) {
- unsigned int snum = csrc->mc_snum;
+ for (i = 0; i < NUMKEYS(psrc); i++, j++) {
+ srcnode = NODEPTR(psrc, i);
+ if (i == 0 && IS_BRANCH(psrc)) {
+ MDB_cursor mn;
MDB_node *s2;
+ mdb_cursor_copy(csrc, &mn);
/* must find the lowest key below src */
- mdb_page_search_lowest(csrc);
- if (IS_LEAF2(csrc->mc_pg[csrc->mc_top])) {
- key.mv_size = csrc->mc_db->md_pad;
- key.mv_data = LEAF2KEY(csrc->mc_pg[csrc->mc_top], 0, key.mv_size);
+ rc = mdb_page_search_lowest(&mn);
+ if (rc)
+ return rc;
+ if (IS_LEAF2(mn.mc_pg[mn.mc_top])) {
+ key.mv_size = mn.mc_db->md_pad;
+ key.mv_data = LEAF2KEY(mn.mc_pg[mn.mc_top], 0, key.mv_size);
} else {
- s2 = NODEPTR(csrc->mc_pg[csrc->mc_top], 0);
+ s2 = NODEPTR(mn.mc_pg[mn.mc_top], 0);
key.mv_size = NODEKSZ(s2);
key.mv_data = NODEKEY(s2);
}
- csrc->mc_snum = snum--;
- csrc->mc_top = snum;
} else {
key.mv_size = srcnode->mn_ksize;
key.mv_data = NODEKEY(srcnode);
}
DPRINTF(("dst page %"Z"u now has %u keys (%.1f%% filled)",
- cdst->mc_pg[cdst->mc_top]->mp_pgno, NUMKEYS(cdst->mc_pg[cdst->mc_top]),
- (float)PAGEFILL(cdst->mc_txn->mt_env, cdst->mc_pg[cdst->mc_top]) / 10));
+ pdst->mp_pgno, NUMKEYS(pdst),
+ (float)PAGEFILL(cdst->mc_txn->mt_env, pdst) / 10));
/* Unlink the src page from parent and add to free list.
*/
}
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--;
/* Adjust other cursors pointing to mp */
MDB_cursor *m2, *m3;
MDB_dbi dbi = csrc->mc_dbi;
- MDB_page *mp = cdst->mc_pg[cdst->mc_top];
for (m2 = csrc->mc_txn->mt_cursors[dbi]; m2; m2=m2->mc_next) {
if (csrc->mc_flags & C_SUB)
m3 = m2;
if (m3 == csrc) continue;
if (m3->mc_snum < csrc->mc_snum) continue;
- if (m3->mc_pg[csrc->mc_top] == csrc->mc_pg[csrc->mc_top]) {
- m3->mc_pg[csrc->mc_top] = mp;
+ if (m3->mc_pg[csrc->mc_top] == psrc) {
+ m3->mc_pg[csrc->mc_top] = pdst;
m3->mc_ki[csrc->mc_top] += nkeys;
}
}
}
- mdb_cursor_pop(csrc);
-
- return mdb_rebalance(csrc);
+ {
+ unsigned int snum = cdst->mc_snum;
+ uint16_t depth = cdst->mc_db->md_depth;
+ mdb_cursor_pop(cdst);
+ rc = mdb_rebalance(cdst);
+ /* Did the tree height change? */
+ if (depth != cdst->mc_db->md_depth)
+ snum += cdst->mc_db->md_depth - depth;
+ cdst->mc_snum = snum;
+ cdst->mc_top = snum-1;
+ }
+ return rc;
}
/** Copy the contents of a cursor.
{
MDB_node *node;
int rc;
- unsigned int ptop, minkeys;
+ unsigned int ptop, minkeys, thresh;
MDB_cursor mn;
+ indx_t oldki;
- minkeys = 1 + (IS_BRANCH(mc->mc_pg[mc->mc_top]));
+ if (IS_BRANCH(mc->mc_pg[mc->mc_top])) {
+ minkeys = 2;
+ thresh = 1;
+ } else {
+ minkeys = 1;
+ thresh = FILL_THRESHOLD;
+ }
DPRINTF(("rebalancing %s page %"Z"u (has %u keys, %.1f%% full)",
IS_LEAF(mc->mc_pg[mc->mc_top]) ? "leaf" : "branch",
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));
- if (PAGEFILL(mc->mc_txn->mt_env, mc->mc_pg[mc->mc_top]) >= FILL_THRESHOLD &&
+ if (PAGEFILL(mc->mc_txn->mt_env, mc->mc_pg[mc->mc_top]) >= thresh &&
NUMKEYS(mc->mc_pg[mc->mc_top]) >= minkeys) {
DPRINTF(("no need to rebalance page %"Z"u, above fill threshold",
mdb_dbg_pgno(mc->mc_pg[mc->mc_top])));
}
}
} else if (IS_BRANCH(mp) && NUMKEYS(mp) == 1) {
+ int i;
DPUTS("collapsing root page!");
rc = mdb_midl_append(&mc->mc_txn->mt_free_pgs, mp->mp_pgno);
if (rc)
mc->mc_db->md_depth--;
mc->mc_db->md_branch_pages--;
mc->mc_ki[0] = mc->mc_ki[1];
+ for (i = 1; i<mc->mc_db->md_depth; i++) {
+ mc->mc_pg[i] = mc->mc_pg[i+1];
+ mc->mc_ki[i] = mc->mc_ki[i+1];
+ }
{
/* Adjust other cursors pointing to mp */
MDB_cursor *m2, *m3;
m3 = m2;
if (m3 == mc || m3->mc_snum < mc->mc_snum) continue;
if (m3->mc_pg[0] == mp) {
- int i;
- m3->mc_snum--;
- m3->mc_top--;
for (i=0; i<m3->mc_snum; i++) {
m3->mc_pg[i] = m3->mc_pg[i+1];
m3->mc_ki[i] = m3->mc_ki[i+1];
}
+ m3->mc_snum--;
+ m3->mc_top--;
}
}
}
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.
*/
* move one key from it. Otherwise we should try to merge them.
* (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]) >= thresh && NUMKEYS(mn.mc_pg[mn.mc_top]) > minkeys) {
+ rc = mdb_node_move(&mn, mc);
+ if (mc->mc_ki[mc->mc_top-1]) {
+ oldki++;
+ }
+ } else {
+ if (mc->mc_ki[ptop] == 0) {
rc = mdb_page_merge(&mn, mc);
- else {
+ } else {
+ MDB_cursor dummy;
+ oldki += NUMKEYS(mn.mc_pg[mn.mc_top]);
mn.mc_ki[mn.mc_top] += mc->mc_ki[mn.mc_top] + 1;
+ /* We want mdb_rebalance to find mn when doing fixups */
+ if (mc->mc_flags & C_SUB) {
+ dummy.mc_next = mc->mc_txn->mt_cursors[mc->mc_dbi];
+ mc->mc_txn->mt_cursors[mc->mc_dbi] = &dummy;
+ dummy.mc_xcursor = (MDB_xcursor *)&mn;
+ } else {
+ mn.mc_next = mc->mc_txn->mt_cursors[mc->mc_dbi];
+ mc->mc_txn->mt_cursors[mc->mc_dbi] = &mn;
+ }
rc = mdb_page_merge(mc, &mn);
+ if (mc->mc_flags & C_SUB)
+ mc->mc_txn->mt_cursors[mc->mc_dbi] = dummy.mc_next;
+ else
+ mc->mc_txn->mt_cursors[mc->mc_dbi] = mn.mc_next;
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;
+ MDB_cursor *m2, *m3;
+ MDB_dbi dbi = mc->mc_dbi;
- 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;
- }
+ mp = mc->mc_pg[mc->mc_top];
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, *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);
-
+ {
/* Adjust other cursors pointing to mp */
for (m2 = mc->mc_txn->mt_cursors[dbi]; m2; m2=m2->mc_next) {
m3 = (mc->mc_flags & C_SUB) ? &m2->mc_xcursor->mx_cursor : m2;
m3->mc_flags |= C_DEL;
if (m3->mc_ki[mc->mc_top] > ki)
m3->mc_ki[mc->mc_top]--;
+ else if (mc->mc_db->md_flags & MDB_DUPSORT)
+ m3->mc_xcursor->mx_cursor.mc_flags |= C_EOF;
+ }
+ }
+ }
+ }
+ rc = mdb_rebalance(mc);
+
+ if (rc == MDB_SUCCESS) {
+ /* DB is totally empty now, just bail out.
+ * Other cursors adjustments were already done
+ * by mdb_rebalance and aren't needed here.
+ */
+ if (!mc->mc_snum)
+ return rc;
+
+ mp = mc->mc_pg[mc->mc_top];
+ nkeys = NUMKEYS(mp);
+
+ /* Adjust other cursors pointing to mp */
+ 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 (m3->mc_snum < mc->mc_snum)
+ continue;
+ if (m3->mc_pg[mc->mc_top] == mp) {
+ /* if m3 points past last node in page, find next sibling */
+ if (m3->mc_ki[mc->mc_top] >= nkeys) {
+ rc = mdb_cursor_sibling(m3, 1);
+ if (rc == MDB_NOTFOUND) {
+ m3->mc_flags |= C_EOF;
+ rc = MDB_SUCCESS;
+ }
}
- if (m3->mc_ki[mc->mc_top] >= nkeys)
- mdb_cursor_sibling(m3, 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 || !TXN_DBI_EXIST(txn, dbi, DB_USRVALID))
+ return EINVAL;
+
+ if (txn->mt_flags & (MDB_TXN_RDONLY|MDB_TXN_BLOCKED))
+ 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;
- if (key == NULL)
- return EINVAL;
-
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;
} else {
op = MDB_SET;
xdata = NULL;
+ flags |= MDB_NODUPDATA;
}
rc = mdb_cursor_set(&mc, key, xdata, op, &exact);
if (rc == 0) {
mc.mc_flags |= C_UNTRACK;
mc.mc_next = txn->mt_cursors[dbi];
txn->mt_cursors[dbi] = &mc;
- rc = mdb_cursor_del(&mc, data ? 0 : MDB_NODUPDATA);
+ rc = mdb_cursor_del(&mc, flags);
txn->mt_cursors[dbi] = mc.mc_next;
}
return rc;
/* Create a right sibling. */
if ((rc = mdb_page_new(mc, mp->mp_flags, 1, &rp)))
return rc;
+ rp->mp_pad = mp->mp_pad;
DPRINTF(("new right sibling: page %"Z"u", rp->mp_pgno));
if (mc->mc_snum < 2) {
if ((rc = mdb_page_new(mc, P_BRANCH, 1, &pp)))
- return rc;
+ goto done;
/* shift current top to make room for new parent */
mc->mc_pg[1] = mc->mc_pg[0];
mc->mc_ki[1] = mc->mc_ki[0];
mc->mc_ki[0] = 0;
mc->mc_db->md_root = pp->mp_pgno;
DPRINTF(("root split! new root = %"Z"u", pp->mp_pgno));
- mc->mc_db->md_depth++;
- new_root = 1;
+ new_root = mc->mc_db->md_depth++;
/* Add left (implicit) pointer. */
if ((rc = mdb_node_add(mc, 0, NULL, NULL, mp->mp_pgno, 0)) != MDB_SUCCESS) {
mc->mc_ki[0] = mc->mc_ki[1];
mc->mc_db->md_root = mp->mp_pgno;
mc->mc_db->md_depth--;
- return rc;
+ goto done;
}
mc->mc_snum = 2;
mc->mc_top = 1;
int x;
unsigned int lsize, rsize, ksize;
/* Move half of the keys to the right sibling */
- copy = NULL;
x = mc->mc_ki[mc->mc_top] - split_indx;
ksize = mc->mc_db->md_pad;
split = LEAF2KEY(mp, split_indx, ksize);
/* 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;
- copy->mp_upper = env->me_psize;
+ copy->mp_lower = (PAGEHDRSZ-PAGEBASE);
+ copy->mp_upper = env->me_psize - PAGEBASE;
/* prepare to insert */
for (i=0, j=0; i<nkeys; i++) {
psize = 0;
if (newindx <= split_indx || newindx >= nkeys) {
i = 0; j = 1;
- k = newindx >= nkeys ? nkeys : split_indx+2;
+ k = newindx >= nkeys ? nkeys : split_indx+1+IS_LEAF(mp);
} else {
i = nkeys; j = -1;
k = split_indx-1;
psize += nsize;
node = NULL;
} else {
- node = (MDB_node *)((char *)mp + copy->mp_ptrs[i]);
+ node = (MDB_node *)((char *)mp + copy->mp_ptrs[i] + PAGEBASE);
psize += NODESIZE + NODEKSZ(node) + sizeof(indx_t);
if (IS_LEAF(mp)) {
if (F_ISSET(node->mn_flags, F_BIGDATA))
sepkey.mv_size = newkey->mv_size;
sepkey.mv_data = newkey->mv_data;
} else {
- node = (MDB_node *)((char *)mp + copy->mp_ptrs[split_indx]);
+ node = (MDB_node *)((char *)mp + copy->mp_ptrs[split_indx] + PAGEBASE);
sepkey.mv_size = node->mn_ksize;
sepkey.mv_data = NODEKEY(node);
}
mn.mc_top--;
did_split = 1;
rc = mdb_page_split(&mn, &sepkey, NULL, rp->mp_pgno, 0);
+ if (rc)
+ goto done;
/* root split? */
if (mn.mc_snum == mc->mc_snum) {
}
mc->mc_flags ^= C_SPLITTING;
if (rc != MDB_SUCCESS) {
- return rc;
+ goto done;
}
if (nflags & MDB_APPEND) {
mc->mc_pg[mc->mc_top] = rp;
mc->mc_ki[mc->mc_top] = 0;
rc = mdb_node_add(mc, 0, newkey, newdata, newpgno, nflags);
if (rc)
- return rc;
+ goto done;
for (i=0; i<mc->mc_top; i++)
mc->mc_ki[i] = mn.mc_ki[i];
} else if (!IS_LEAF2(mp)) {
/* Update index for the new key. */
mc->mc_ki[mc->mc_top] = j;
} else {
- node = (MDB_node *)((char *)mp + copy->mp_ptrs[i]);
+ node = (MDB_node *)((char *)mp + copy->mp_ptrs[i] + PAGEBASE);
rkey.mv_data = NODEKEY(node);
rkey.mv_size = node->mn_ksize;
if (IS_LEAF(mp)) {
}
rc = mdb_node_add(mc, j, &rkey, rdata, pgno, flags);
- if (rc) {
- /* return tmp page to freelist */
- mdb_page_free(env, copy);
- return rc;
- }
+ if (rc)
+ goto done;
if (i == nkeys) {
i = 0;
j = 0;
mp->mp_lower = copy->mp_lower;
mp->mp_upper = copy->mp_upper;
memcpy(NODEPTR(mp, nkeys-1), NODEPTR(copy, nkeys-1),
- env->me_psize - copy->mp_upper);
+ env->me_psize - copy->mp_upper - PAGEBASE);
/* reset back to original page */
if (newindx < split_indx) {
}
}
}
- /* return tmp page to freelist */
- mdb_page_free(env, copy);
}
{
if (new_root) {
int k;
/* root split */
- for (k=m3->mc_top; k>=0; k--) {
+ for (k=new_root; k>=0; k--) {
m3->mc_ki[k+1] = m3->mc_ki[k];
m3->mc_pg[k+1] = m3->mc_pg[k];
}
}
}
DPRINTF(("mp left: %d, rp left: %d", SIZELEFT(mp), SIZELEFT(rp)));
+
+done:
+ if (copy) /* tmp page */
+ mdb_page_free(env, copy);
+ if (rc)
+ mc->mc_txn->mt_flags |= MDB_TXN_ERROR;
return rc;
}
MDB_cursor mc;
MDB_xcursor mx;
- if (key == NULL || data == NULL)
+ if (!key || !data || !TXN_DBI_EXIST(txn, dbi, DB_USRVALID))
return EINVAL;
- if (txn == NULL || !dbi || dbi >= txn->mt_numdbs || !(txn->mt_dbflags[dbi] & DB_VALID))
+ if (flags & ~(MDB_NOOVERWRITE|MDB_NODUPDATA|MDB_RESERVE|MDB_APPEND|MDB_APPENDDUP))
return EINVAL;
- if ((flags & (MDB_NOOVERWRITE|MDB_NODUPDATA|MDB_RESERVE|MDB_APPEND|MDB_APPENDDUP)) != flags)
- return EINVAL;
+ if (txn->mt_flags & (MDB_TXN_RDONLY|MDB_TXN_BLOCKED))
+ return (txn->mt_flags & MDB_TXN_RDONLY) ? EACCES : MDB_BAD_TXN;
mdb_cursor_init(&mc, txn, dbi, &mx);
return mdb_cursor_put(&mc, key, data, flags);
}
-int
-mdb_env_set_flags(MDB_env *env, unsigned int flag, int onoff)
-{
- if ((flag & CHANGEABLE) != flag)
- return EINVAL;
- if (onoff)
- env->me_flags |= flag;
- else
- env->me_flags &= ~flag;
- return MDB_SUCCESS;
-}
-
-int
-mdb_env_get_flags(MDB_env *env, unsigned int *arg)
-{
- if (!env || !arg)
- return EINVAL;
+#ifndef MDB_WBUF
+#define MDB_WBUF (1024*1024)
+#endif
- *arg = env->me_flags;
- return MDB_SUCCESS;
-}
+ /** 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 ESECT CALL_CONV
+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
-int
-mdb_env_set_userctx(MDB_env *env, void *ctx)
-{
- if (!env)
- return EINVAL;
+ 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 ESECT
+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 ESECT
+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; i<mc.mc_top; i++) {
+ mdb_page_copy((MDB_page *)ptr, mc.mc_pg[i], my->mc_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; i<n; i++) {
+ ni = NODEPTR(mp, i);
+ if (ni->mn_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 ESECT
+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_os_psize);
+ if (my.mc_wbuf[0] == NULL)
+ return errno;
+#else
+ pthread_mutex_init(&my.mc_mutex, NULL);
+ pthread_cond_init(&my.mc_cond, NULL);
+#ifdef HAVE_MEMALIGN
+ my.mc_wbuf[0] = memalign(env->me_os_psize, MDB_WBUF*2);
+ if (my.mc_wbuf[0] == NULL)
+ return errno;
+#else
+ rc = posix_memalign((void **)&my.mc_wbuf[0], env->me_os_psize, MDB_WBUF*2);
+ if (rc)
+ return rc;
+#endif
+#endif
+ memset(my.mc_wbuf[0], 0, MDB_WBUF*2);
+ 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 = NUM_METAS;
+ my.mc_status = 0;
+ my.mc_new = 1;
+ my.mc_toggle = 0;
+ my.mc_env = env;
+ my.mc_fd = fd;
+ THREAD_CREATE(thr, mdb_env_copythr, &my);
+
+ rc = mdb_txn_begin(env, NULL, MDB_RDONLY, &txn);
+ if (rc)
+ return rc;
+
+ mp = (MDB_page *)my.mc_wbuf[0];
+ memset(mp, 0, NUM_METAS * 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[FREE_DBI].md_branch_pages +
+ txn->mt_dbs[FREE_DBI].md_leaf_pages +
+ txn->mt_dbs[FREE_DBI].md_overflow_pages;
+
+ /* Set metapage 1 */
+ mm->mm_last_pg = txn->mt_next_pgno - freecount - 1;
+ mm->mm_dbs[MAIN_DBI] = txn->mt_dbs[MAIN_DBI];
+ if (mm->mm_last_pg > NUM_METAS-1) {
+ mm->mm_dbs[MAIN_DBI].md_root = mm->mm_last_pg;
+ mm->mm_txnid = 1;
+ } else {
+ mm->mm_dbs[MAIN_DBI].md_root = P_INVALID;
+ }
+ }
+ my.mc_wlen[0] = env->me_psize * NUM_METAS;
+ 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[MAIN_DBI].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);
+
+ 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 ESECT
+mdb_env_copyfd0(MDB_env *env, HANDLE fd)
+{
+ MDB_txn *txn = NULL;
+ mdb_mutexref_t wmutex = 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_end(txn, MDB_END_RESET_TMP);
+
+ /* Temporarily block writers until we snapshot the meta pages */
+ wmutex = env->me_wmutex;
+ if (LOCK_MUTEX(rc, env, wmutex))
+ goto leave;
+
+ rc = mdb_txn_renew0(txn);
+ if (rc) {
+ UNLOCK_MUTEX(wmutex);
+ goto leave;
+ }
+ }
+
+ wsize = env->me_psize * NUM_METAS;
+ 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 (wmutex)
+ UNLOCK_MUTEX(wmutex);
+
+ if (rc)
+ goto leave;
+
+ w2 = txn->mt_next_pgno * env->me_psize;
+ {
+ size_t fsize = 0;
+ if ((rc = mdb_fsize(env->me_fd, &fsize)))
+ goto leave;
+ if (w2 > fsize)
+ w2 = fsize;
+ }
+ 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 ESECT
+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 ESECT
+mdb_env_copyfd(MDB_env *env, HANDLE fd)
+{
+ return mdb_env_copyfd2(env, fd, 0);
+}
+
+int ESECT
+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;
+ }
+
+ if (env->me_psize >= env->me_os_psize) {
+#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 ESECT
+mdb_env_copy(MDB_env *env, const char *path)
+{
+ return mdb_env_copy2(env, path, 0);
+}
+
+int ESECT
+mdb_env_set_flags(MDB_env *env, unsigned int flag, int onoff)
+{
+ if (flag & ~CHANGEABLE)
+ return EINVAL;
+ if (onoff)
+ env->me_flags |= flag;
+ else
+ env->me_flags &= ~flag;
+ return MDB_SUCCESS;
+}
+
+int ESECT
+mdb_env_get_flags(MDB_env *env, unsigned int *arg)
+{
+ if (!env || !arg)
+ return EINVAL;
+
+ *arg = env->me_flags & (CHANGEABLE|CHANGELESS);
+ return MDB_SUCCESS;
+}
+
+int ESECT
+mdb_env_set_userctx(MDB_env *env, void *ctx)
+{
+ if (!env)
+ return EINVAL;
env->me_userctx = ctx;
return MDB_SUCCESS;
}
-void *
+void * ESECT
mdb_env_get_userctx(MDB_env *env)
{
return env ? env->me_userctx : NULL;
}
-int
+int ESECT
mdb_env_set_assert(MDB_env *env, MDB_assert_func *func)
{
if (!env)
return MDB_SUCCESS;
}
-int
+int ESECT
mdb_env_get_path(MDB_env *env, const char **arg)
{
if (!env || !arg)
return MDB_SUCCESS;
}
-int
+int ESECT
mdb_env_get_fd(MDB_env *env, mdb_filehandle_t *arg)
{
if (!env || !arg)
* @param[out] arg the address of an #MDB_stat structure to receive the stats.
* @return 0, this function always succeeds.
*/
-static int
+static int ESECT
mdb_stat0(MDB_env *env, MDB_db *db, MDB_stat *arg)
{
arg->ms_psize = env->me_psize;
return MDB_SUCCESS;
}
-int
+
+int ESECT
mdb_env_stat(MDB_env *env, MDB_stat *arg)
{
- int toggle;
+ MDB_meta *meta;
if (env == NULL || arg == NULL)
return EINVAL;
- toggle = mdb_env_pick_meta(env);
+ meta = mdb_env_pick_meta(env);
- return mdb_stat0(env, &env->me_metas[toggle]->mm_dbs[MAIN_DBI], arg);
+ return mdb_stat0(env, &meta->mm_dbs[MAIN_DBI], arg);
}
-int
+int ESECT
mdb_env_info(MDB_env *env, MDB_envinfo *arg)
{
- int toggle;
+ MDB_meta *meta;
if (env == NULL || arg == NULL)
return EINVAL;
- toggle = mdb_env_pick_meta(env);
- arg->me_mapaddr = (env->me_flags & MDB_FIXEDMAP) ? env->me_map : 0;
+ meta = mdb_env_pick_meta(env);
+ arg->me_mapaddr = meta->mm_address;
+ arg->me_last_pgno = meta->mm_last_pg;
+ arg->me_last_txnid = meta->mm_txnid;
+
arg->me_mapsize = env->me_mapsize;
arg->me_maxreaders = env->me_maxreaders;
-
- /* me_numreaders may be zero if this process never used any readers. Use
- * the shared numreader count if it exists.
- */
- arg->me_numreaders = env->me_txns ? env->me_txns->mti_numreaders : env->me_numreaders;
-
- arg->me_last_pgno = env->me_metas[toggle]->mm_last_pg;
- arg->me_last_txnid = env->me_metas[toggle]->mm_txnid;
+ arg->me_numreaders = env->me_txns ? env->me_txns->mti_numreaders : 0;
return MDB_SUCCESS;
}
MDB_val key, data;
MDB_dbi i;
MDB_cursor mc;
+ MDB_db dummy;
int rc, dbflag, exact;
- unsigned int unused = 0;
+ unsigned int unused = 0, seq;
size_t len;
- if (txn->mt_dbxs[FREE_DBI].md_cmp == NULL) {
- mdb_default_cmp(txn, FREE_DBI);
- }
-
- if ((flags & VALID_FLAGS) != flags)
+ if (flags & ~VALID_FLAGS)
return EINVAL;
- if (txn->mt_flags & MDB_TXN_ERROR)
+ if (txn->mt_flags & MDB_TXN_BLOCKED)
return MDB_BAD_TXN;
/* main DB? */
/* Is the DB already open? */
len = strlen(name);
- for (i=2; i<txn->mt_numdbs; i++) {
+ for (i=CORE_DBS; i<txn->mt_numdbs; i++) {
if (!txn->mt_dbxs[i].md_name.mv_size) {
/* Remember this free slot */
if (!unused) unused = i;
return (flags & MDB_CREATE) ? MDB_INCOMPATIBLE : MDB_NOTFOUND;
/* Find the DB info */
- dbflag = DB_NEW|DB_VALID;
+ dbflag = DB_NEW|DB_VALID|DB_USRVALID;
exact = 0;
key.mv_size = len;
key.mv_data = (void *)name;
if (rc == MDB_SUCCESS) {
/* make sure this is actually a DB */
MDB_node *node = NODEPTR(mc.mc_pg[mc.mc_top], mc.mc_ki[mc.mc_top]);
- if (!(node->mn_flags & F_SUBDATA))
+ if ((node->mn_flags & (F_DUPDATA|F_SUBDATA)) != F_SUBDATA)
return MDB_INCOMPATIBLE;
} else if (rc == MDB_NOTFOUND && (flags & MDB_CREATE)) {
/* Create if requested */
- MDB_db dummy;
data.mv_size = sizeof(MDB_db);
data.mv_data = &dummy;
memset(&dummy, 0, sizeof(dummy));
txn->mt_dbxs[slot].md_name.mv_size = len;
txn->mt_dbxs[slot].md_rel = NULL;
txn->mt_dbflags[slot] = dbflag;
+ /* txn-> and env-> are the same in read txns, use
+ * tmp variable to avoid undefined assignment
+ */
+ seq = ++txn->mt_env->me_dbiseqs[slot];
+ txn->mt_dbiseqs[slot] = seq;
+
memcpy(&txn->mt_dbs[slot], data.mv_data, sizeof(MDB_db));
*dbi = slot;
mdb_default_cmp(txn, slot);
return rc;
}
-int mdb_stat(MDB_txn *txn, MDB_dbi dbi, MDB_stat *arg)
+int ESECT
+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, DB_VALID))
return EINVAL;
+ if (txn->mt_flags & MDB_TXN_BLOCKED)
+ return MDB_BAD_TXN;
+
if (txn->mt_dbflags[dbi] & DB_STALE) {
MDB_cursor mc;
MDB_xcursor mx;
void mdb_dbi_close(MDB_env *env, MDB_dbi dbi)
{
char *ptr;
- if (dbi <= MAIN_DBI || dbi >= env->me_maxdbs)
+ if (dbi < CORE_DBS || dbi >= env->me_maxdbs)
return;
ptr = env->me_dbxs[dbi].md_name.mv_data;
- env->me_dbxs[dbi].md_name.mv_data = NULL;
- env->me_dbxs[dbi].md_name.mv_size = 0;
- env->me_dbflags[dbi] = 0;
- free(ptr);
+ /* If there was no name, this was already closed */
+ if (ptr) {
+ env->me_dbxs[dbi].md_name.mv_data = NULL;
+ env->me_dbxs[dbi].md_name.mv_size = 0;
+ env->me_dbflags[dbi] = 0;
+ env->me_dbiseqs[dbi]++;
+ free(ptr);
+ }
}
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 (!TXN_DBI_EXIST(txn, dbi, DB_USRVALID))
return EINVAL;
*flags = txn->mt_dbs[dbi].md_flags & PERSISTENT_FLAGS;
return MDB_SUCCESS;
MDB_cursor mx;
unsigned int i;
- /* LEAF2 pages have no nodes, cannot have sub-DBs */
- if (IS_LEAF2(mc->mc_pg[mc->mc_top]))
+ /* DUPSORT sub-DBs have no ovpages/DBs. Omit scanning leaves.
+ * This also avoids any P_LEAF2 pages, which have no nodes.
+ */
+ if (mc->mc_flags & C_SUB)
mdb_cursor_pop(mc);
mdb_cursor_copy(mc, &mx);
memcpy(&pg, NODEDATA(ni), sizeof(pg));
rc = mdb_page_get(txn, pg, &omp, NULL);
if (rc != 0)
- return rc;
+ goto done;
mdb_cassert(mc, IS_OVERFLOW(omp));
rc = mdb_midl_append_range(&txn->mt_free_pgs,
pg, omp->mp_pages);
if (rc)
- return rc;
+ goto done;
} else if (subs && (ni->mn_flags & F_SUBDATA)) {
mdb_xcursor_init1(mc, ni);
rc = mdb_drop0(&mc->mc_xcursor->mx_cursor, 0);
if (rc)
- return rc;
+ goto done;
}
}
} else {
if ((rc = mdb_midl_need(&txn->mt_free_pgs, n)) != 0)
- return rc;
+ goto done;
for (i=0; i<n; i++) {
pgno_t pg;
ni = NODEPTR(mp, i);
mc->mc_ki[mc->mc_top] = i;
rc = mdb_cursor_sibling(mc, 1);
if (rc) {
+ if (rc != MDB_NOTFOUND)
+ goto done;
/* no more siblings, go back to beginning
* of previous level.
*/
}
/* free it */
rc = mdb_midl_append(&txn->mt_free_pgs, mc->mc_db->md_root);
+done:
+ if (rc)
+ txn->mt_flags |= MDB_TXN_ERROR;
} else if (rc == MDB_NOTFOUND) {
rc = MDB_SUCCESS;
}
MDB_cursor *mc, *m2;
int rc;
- if (!txn || !dbi || dbi >= txn->mt_numdbs || (unsigned)del > 1 || !(txn->mt_dbflags[dbi] & DB_VALID))
+ if ((unsigned)del > 1 || !TXN_DBI_EXIST(txn, dbi, DB_USRVALID))
return EINVAL;
if (F_ISSET(txn->mt_flags, MDB_TXN_RDONLY))
return EACCES;
+ if (TXN_DBI_CHANGED(txn, dbi))
+ return MDB_BAD_DBI;
+
rc = mdb_cursor_open(txn, dbi, &mc);
if (rc)
return rc;
goto leave;
/* Can't delete the main DB */
- if (del && dbi > MAIN_DBI) {
- rc = mdb_del(txn, MAIN_DBI, &mc->mc_dbx->md_name, NULL);
+ if (del && dbi >= CORE_DBS) {
+ rc = mdb_del0(txn, MAIN_DBI, &mc->mc_dbx->md_name, NULL, F_SUBDATA);
if (!rc) {
txn->mt_dbflags[dbi] = DB_STALE;
mdb_dbi_close(txn->mt_env, dbi);
+ } else {
+ txn->mt_flags |= MDB_TXN_ERROR;
}
} else {
/* reset the DB record, mark it dirty */
int mdb_set_compare(MDB_txn *txn, MDB_dbi dbi, MDB_cmp_func *cmp)
{
- if (txn == NULL || !dbi || dbi >= txn->mt_numdbs || !(txn->mt_dbflags[dbi] & DB_VALID))
+ if (!TXN_DBI_EXIST(txn, dbi, DB_USRVALID))
return EINVAL;
txn->mt_dbxs[dbi].md_cmp = cmp;
int mdb_set_dupsort(MDB_txn *txn, MDB_dbi dbi, MDB_cmp_func *cmp)
{
- if (txn == NULL || !dbi || dbi >= txn->mt_numdbs || !(txn->mt_dbflags[dbi] & DB_VALID))
+ if (!TXN_DBI_EXIST(txn, dbi, DB_USRVALID))
return EINVAL;
txn->mt_dbxs[dbi].md_dcmp = cmp;
int mdb_set_relfunc(MDB_txn *txn, MDB_dbi dbi, MDB_rel_func *rel)
{
- if (txn == NULL || !dbi || dbi >= txn->mt_numdbs || !(txn->mt_dbflags[dbi] & DB_VALID))
+ if (!TXN_DBI_EXIST(txn, dbi, DB_USRVALID))
return EINVAL;
txn->mt_dbxs[dbi].md_rel = rel;
int mdb_set_relctx(MDB_txn *txn, MDB_dbi dbi, void *ctx)
{
- if (txn == NULL || !dbi || dbi >= txn->mt_numdbs || !(txn->mt_dbflags[dbi] & DB_VALID))
+ if (!TXN_DBI_EXIST(txn, dbi, DB_USRVALID))
return EINVAL;
txn->mt_dbxs[dbi].md_relctx = ctx;
return MDB_SUCCESS;
}
-int mdb_env_get_maxkeysize(MDB_env *env)
+int ESECT
+mdb_env_get_maxkeysize(MDB_env *env)
{
return ENV_MAXKEY(env);
}
-int mdb_reader_list(MDB_env *env, MDB_msg_func *func, void *ctx)
+int ESECT
+mdb_reader_list(MDB_env *env, MDB_msg_func *func, void *ctx)
{
unsigned int i, rdrs;
MDB_reader *mr;
/** Insert pid into list if not already present.
* return -1 if already present.
*/
-static int mdb_pid_insert(MDB_PID_T *ids, MDB_PID_T pid)
+static int ESECT
+mdb_pid_insert(MDB_PID_T *ids, MDB_PID_T pid)
{
/* binary search of pid in list */
unsigned base = 0;
return 0;
}
-int mdb_reader_check(MDB_env *env, int *dead)
+int ESECT
+mdb_reader_check(MDB_env *env, int *dead)
{
- unsigned int i, j, rdrs;
- MDB_reader *mr;
- MDB_PID_T *pids, pid;
- int count = 0;
-
if (!env)
return EINVAL;
if (dead)
*dead = 0;
- if (!env->me_txns)
- return MDB_SUCCESS;
+ return env->me_txns ? mdb_reader_check0(env, 0, dead) : MDB_SUCCESS;
+}
+
+/** As #mdb_reader_check(). rlocked = <caller locked the reader mutex>. */
+static int ESECT
+mdb_reader_check0(MDB_env *env, int rlocked, int *dead)
+{
+ mdb_mutexref_t rmutex = rlocked ? NULL : env->me_rmutex;
+ unsigned int i, j, rdrs;
+ MDB_reader *mr;
+ MDB_PID_T *pids, pid;
+ int rc = MDB_SUCCESS, count = 0;
+
rdrs = env->me_txns->mti_numreaders;
pids = malloc((rdrs+1) * sizeof(MDB_PID_T));
if (!pids)
pids[0] = 0;
mr = env->me_txns->mti_readers;
for (i=0; i<rdrs; i++) {
- if (mr[i].mr_pid && mr[i].mr_pid != env->me_pid) {
- pid = mr[i].mr_pid;
+ pid = mr[i].mr_pid;
+ if (pid && pid != env->me_pid) {
if (mdb_pid_insert(pids, pid) == 0) {
if (!mdb_reader_pid(env, Pidcheck, pid)) {
- LOCK_MUTEX_R(env);
- /* Recheck, a new process may have reused pid */
- if (!mdb_reader_pid(env, Pidcheck, pid)) {
- for (j=i; j<rdrs; j++)
+ /* Stale reader found */
+ j = i;
+ if (rmutex) {
+ if ((rc = LOCK_MUTEX0(rmutex)) != 0) {
+ if ((rc = mdb_mutex_failed(env, rmutex, rc)))
+ break;
+ rdrs = 0; /* the above checked all readers */
+ } else {
+ /* Recheck, a new process may have reused pid */
+ if (mdb_reader_pid(env, Pidcheck, pid))
+ j = rdrs;
+ }
+ }
+ for (; j<rdrs; j++)
if (mr[j].mr_pid == pid) {
DPRINTF(("clear stale reader pid %u txn %"Z"d",
(unsigned) pid, mr[j].mr_txnid));
mr[j].mr_pid = 0;
count++;
}
- }
- UNLOCK_MUTEX_R(env);
+ if (rmutex)
+ UNLOCK_MUTEX(rmutex);
}
}
}
free(pids);
if (dead)
*dead = count;
- return MDB_SUCCESS;
+ return rc;
+}
+
+#ifdef MDB_ROBUST_SUPPORTED
+/** Handle #LOCK_MUTEX0() failure.
+ * Try to repair the lock file if the mutex owner died.
+ * @param[in] env the environment handle
+ * @param[in] mutex LOCK_MUTEX0() mutex
+ * @param[in] rc LOCK_MUTEX0() error (nonzero)
+ * @return 0 on success with the mutex locked, or an error code on failure.
+ */
+static int ESECT
+mdb_mutex_failed(MDB_env *env, mdb_mutexref_t mutex, int rc)
+{
+ int rlocked, rc2;
+ MDB_meta *meta;
+
+ if (rc == MDB_OWNERDEAD) {
+ /* We own the mutex. Clean up after dead previous owner. */
+ rc = MDB_SUCCESS;
+ rlocked = (mutex == env->me_rmutex);
+ if (!rlocked) {
+ /* Keep mti_txnid updated, otherwise next writer can
+ * overwrite data which latest meta page refers to.
+ */
+ meta = mdb_env_pick_meta(env);
+ env->me_txns->mti_txnid = meta->mm_txnid;
+ /* env is hosed if the dead thread was ours */
+ if (env->me_txn) {
+ env->me_flags |= MDB_FATAL_ERROR;
+ env->me_txn = NULL;
+ rc = MDB_PANIC;
+ }
+ }
+ DPRINTF(("%cmutex owner died, %s", (rlocked ? 'r' : 'w'),
+ (rc ? "this process' env is hosed" : "recovering")));
+ rc2 = mdb_reader_check0(env, rlocked, NULL);
+ if (rc2 == 0)
+ rc2 = mdb_mutex_consistent(mutex);
+ if (rc || (rc = rc2)) {
+ DPRINTF(("LOCK_MUTEX recovery failed, %s", mdb_strerror(rc)));
+ UNLOCK_MUTEX(mutex);
+ }
+ } else {
+#ifdef _WIN32
+ rc = ErrCode();
+#endif
+ DPRINTF(("LOCK_MUTEX failed, %s", mdb_strerror(rc)));
+ }
+
+ return rc;
}
+#endif /* MDB_ROBUST_SUPPORTED */
/** @} */