X-Git-Url: https://git.sur5r.net/?a=blobdiff_plain;ds=sidebyside;f=libraries%2Flibmdb%2Fmdb.c;h=6eff56cdbfc7aacfd7072cdb7991126d249b2f82;hb=0952cfdf84d926da5e6ba984fa55b3ab5a2ef9aa;hp=e87fb98d6916948bd1938ec00bde8745411727ea;hpb=e222506b88bcadff3ff63c055e0257233721606e;p=openldap diff --git a/libraries/libmdb/mdb.c b/libraries/libmdb/mdb.c index e87fb98d69..6eff56cdbf 100644 --- a/libraries/libmdb/mdb.c +++ b/libraries/libmdb/mdb.c @@ -5,7 +5,7 @@ * BerkeleyDB API, but much simplified. */ /* - * Copyright 2011 Howard Chu, Symas Corp. + * Copyright 2011-2012 Howard Chu, Symas Corp. * All rights reserved. * * Redistribution and use in source and binary forms, with or without @@ -50,22 +50,70 @@ #include #include #include -#include +#include #include #include #include #include #include +#if !(defined(BYTE_ORDER) || defined(__BYTE_ORDER)) +#include /* defines BYTE_ORDER on HPUX and Solaris */ +#endif + #ifndef _WIN32 #include +#ifdef __APPLE__ +#include +#endif +#endif + +#ifdef USE_VALGRIND +#include +#define VGMEMP_CREATE(h,r,z) VALGRIND_CREATE_MEMPOOL(h,r,z) +#define VGMEMP_ALLOC(h,a,s) VALGRIND_MEMPOOL_ALLOC(h,a,s) +#define VGMEMP_FREE(h,a) VALGRIND_MEMPOOL_FREE(h,a) +#define VGMEMP_DESTROY(h) VALGRIND_DESTROY_MEMPOOL(h) +#define VGMEMP_DEFINED(a,s) VALGRIND_MAKE_MEM_DEFINED(a,s) +#else +#define VGMEMP_CREATE(h,r,z) +#define VGMEMP_ALLOC(h,a,s) +#define VGMEMP_FREE(h,a) +#define VGMEMP_DESTROY(h) +#define VGMEMP_DEFINED(a,s) +#endif + +#ifndef BYTE_ORDER +# if (defined(_LITTLE_ENDIAN) || defined(_BIG_ENDIAN)) && !(defined(_LITTLE_ENDIAN) && defined(_BIG_ENDIAN)) +/* Solaris just defines one or the other */ +# define LITTLE_ENDIAN 1234 +# define BIG_ENDIAN 4321 +# ifdef _LITTLE_ENDIAN +# define BYTE_ORDER LITTLE_ENDIAN +# else +# define BYTE_ORDER BIG_ENDIAN +# endif +# else +# define BYTE_ORDER __BYTE_ORDER +# endif +#endif + +#ifndef LITTLE_ENDIAN +#define LITTLE_ENDIAN __LITTLE_ENDIAN +#endif +#ifndef BIG_ENDIAN +#define BIG_ENDIAN __BIG_ENDIAN +#endif + +#if defined(__i386) || defined(__x86_64) +#define MISALIGNED_OK 1 #endif #include "mdb.h" #include "midl.h" -#if (__BYTE_ORDER == __LITTLE_ENDIAN) == (__BYTE_ORDER == __BIG_ENDIAN) -# error "Unknown or unsupported endianness (__BYTE_ORDER)" +#if (BYTE_ORDER == LITTLE_ENDIAN) == (BYTE_ORDER == BIG_ENDIAN) +# error "Unknown or unsupported endianness (BYTE_ORDER)" #elif (-6 & 5) || CHAR_BIT != 8 || UINT_MAX < 0xffffffff || ULONG_MAX % 0xFFFF # error "Two's complement, reasonably sized integer types, please" #endif @@ -102,6 +150,16 @@ #define close(fd) CloseHandle(fd) #define munmap(ptr,len) UnmapViewOfFile(ptr) #else +#ifdef __APPLE__ +#define LOCK_MUTEX_R(env) sem_wait((env)->me_rmutex) +#define UNLOCK_MUTEX_R(env) sem_post((env)->me_rmutex) +#define LOCK_MUTEX_W(env) sem_wait((env)->me_wmutex) +#define UNLOCK_MUTEX_W(env) sem_post((env)->me_wmutex) +#define fdatasync(fd) fsync(fd) +#else +#ifdef ANDROID +#define fdatasync(fd) fsync(fd) +#endif /** Lock the reader mutex. */ #define LOCK_MUTEX_R(env) pthread_mutex_lock(&(env)->me_txns->mti_mutex) @@ -117,6 +175,7 @@ /** Unlock the writer mutex. */ #define UNLOCK_MUTEX_W(env) pthread_mutex_unlock(&(env)->me_txns->mti_wmutex) +#endif /* __APPLE__ */ /** Get the error code for the last failed system function. */ @@ -141,6 +200,12 @@ #define GET_PAGESIZE(x) ((x) = sysconf(_SC_PAGE_SIZE)) #endif +#if defined(_WIN32) || defined(__APPLE__) +#define MNAME_LEN 32 +#else +#define MNAME_LEN (sizeof(pthread_mutex_t)) +#endif + /** @} */ #ifndef _WIN32 @@ -156,6 +221,13 @@ #ifndef MDB_DSYNC # define MDB_DSYNC O_DSYNC #endif +#endif + +/** Function for flushing the data of a file. Define this to fsync + * if fdatasync() is not supported. + */ +#ifndef MDB_FDATASYNC +# define MDB_FDATASYNC fdatasync #endif /** A page number in the database. @@ -176,17 +248,17 @@ typedef ID txnid_t; /** @defgroup debug Debug Macros * @{ */ -#ifndef DEBUG +#ifndef MDB_DEBUG /** Enable debug output. * Set this to 1 for copious tracing. Set to 2 to add dumps of all IDLs * read from and written to the database (used for free space management). */ -#define DEBUG 0 +#define MDB_DEBUG 0 #endif #if !(__STDC_VERSION__ >= 199901L || defined(__GNUC__)) # define DPRINTF (void) /* Vararg macros may be unsupported */ -#elif DEBUG +#elif MDB_DEBUG /** Print a debug message with printf formatting. */ # define DPRINTF(fmt, ...) /**< Requires 2 or more args */ \ fprintf(stderr, "%s:%d " fmt "\n", __func__, __LINE__, __VA_ARGS__) @@ -211,7 +283,7 @@ typedef ID txnid_t; * pressure from other processes is high. So until OSs have * actual paging support for Huge pages, they're not viable. */ -#define PAGESIZE 4096 +#define MDB_PAGESIZE 4096 /** The minimum number of keys required in a database page. * Setting this to a larger value will place a smaller bound on the @@ -239,11 +311,11 @@ typedef ID txnid_t; /** The maximum size of a key in the database. * While data items have essentially unbounded size, we require that * keys all fit onto a regular page. This limit could be raised a bit - * further if needed; to something just under #PAGESIZE / #MDB_MINKEYS. + * further if needed; to something just under #MDB_PAGESIZE / #MDB_MINKEYS. */ #define MAXKEYSIZE 511 -#if DEBUG +#if MDB_DEBUG /** A key buffer. * @ingroup debug * This is used for printing a hex dump of a key's contents. @@ -256,11 +328,11 @@ typedef ID txnid_t; #define DKEY(x) mdb_dkey(x, kbuf) #else #define DKBUF typedef int dummy_kbuf /* so we can put ';' after */ -#define DKEY(x) +#define DKEY(x) 0 #endif /** @defgroup lazylock Lazy Locking - * Macros for locks that are't actually needed. + * Macros for locks that aren't actually needed. * The DB view is always consistent because all writes are wrapped in * the wmutex. Finer-grained locks aren't necessary. * @{ @@ -414,18 +486,21 @@ typedef struct MDB_reader { * For POSIX the actual mutexes reside in the shared memory of this * mapped file. On Windows, mutexes are named objects allocated by the * kernel; we store the mutex names in this mapped file so that other - * processes can grab them. This same approach will also be used on + * processes can grab them. This same approach is also used on * MacOSX/Darwin (using named semaphores) since MacOSX doesn't support - * process-shared POSIX mutexes. + * process-shared POSIX mutexes. For these cases where a named object + * is used, the object name is derived from a 64 bit FNV hash of the + * environment pathname. As such, naming collisions are extremely + * unlikely. If a collision occurs, the results are unpredictable. */ typedef struct MDB_txbody { - /** Stamp identifying this as an MDB lock file. It must be set + /** Stamp identifying this as an MDB file. It must be set * to #MDB_MAGIC. */ uint32_t mtb_magic; /** Version number of this lock file. Must be set to #MDB_VERSION. */ uint32_t mtb_version; -#ifdef _WIN32 - char mtb_rmname[32]; +#if defined(_WIN32) || defined(__APPLE__) + char mtb_rmname[MNAME_LEN]; #else /** Mutex protecting access to this table. * This is the reader lock that #LOCK_MUTEX_R acquires. @@ -463,14 +538,14 @@ typedef struct MDB_txninfo { char pad[(sizeof(MDB_txbody)+CACHELINE-1) & ~(CACHELINE-1)]; } mt1; union { -#ifdef _WIN32 - char mt2_wmname[32]; +#if defined(_WIN32) || defined(__APPLE__) + char mt2_wmname[MNAME_LEN]; #define mti_wmname mt2.mt2_wmname #else pthread_mutex_t mt2_wmutex; #define mti_wmutex mt2.mt2_wmutex #endif - char pad[(sizeof(pthread_mutex_t)+CACHELINE-1) & ~(CACHELINE-1)]; + char pad[(MNAME_LEN+CACHELINE-1) & ~(CACHELINE-1)]; } mt2; MDB_reader mti_readers[1]; } MDB_txninfo; @@ -487,6 +562,7 @@ typedef struct MDB_page { pgno_t p_pgno; /**< page number */ void * p_next; /**< for in-memory list of freed structs */ } mp_p; + uint16_t mp_pad; /** @defgroup mdb_page Page Flags * @ingroup internal * Flags for the page headers. @@ -498,8 +574,9 @@ typedef struct MDB_page { #define P_META 0x08 /**< meta page */ #define P_DIRTY 0x10 /**< dirty page */ #define P_LEAF2 0x20 /**< for #MDB_DUPFIXED records */ +#define P_SUBP 0x40 /**< for #MDB_DUPSORT sub-pages */ /** @} */ - uint32_t mp_flags; /**< @ref mdb_page */ + uint16_t mp_flags; /**< @ref mdb_page */ #define mp_lower mp_pb.pb.pb_lower #define mp_upper mp_pb.pb.pb_upper #define mp_pages mp_pb.pb_pages @@ -541,6 +618,8 @@ typedef struct MDB_page { #define IS_BRANCH(p) F_ISSET((p)->mp_flags, P_BRANCH) /** Test if a page is an overflow page */ #define IS_OVERFLOW(p) F_ISSET((p)->mp_flags, P_OVERFLOW) + /** Test if a page is a sub page */ +#define IS_SUBP(p) F_ISSET((p)->mp_flags, P_SUBP) /** The number of overflow pages needed to store the given size. */ #define OVPAGES(size, psize) ((PAGEHDRSZ-1 + (size)) / (psize) + 1) @@ -555,8 +634,8 @@ typedef struct MDB_node { * They are in host byte order in case that lets some * accesses be optimized into a 32-bit word access. */ -#define mn_lo mn_offset[__BYTE_ORDER!=__LITTLE_ENDIAN] -#define mn_hi mn_offset[__BYTE_ORDER==__LITTLE_ENDIAN] /**< part of dsize or pgno */ +#define mn_lo mn_offset[BYTE_ORDER!=LITTLE_ENDIAN] +#define mn_hi mn_offset[BYTE_ORDER==LITTLE_ENDIAN] /**< part of dsize or pgno */ unsigned short mn_offset[2]; /**< storage for #mn_lo and #mn_hi */ /** @defgroup mdb_node Node Flags * @ingroup internal @@ -566,6 +645,10 @@ typedef struct MDB_node { #define F_BIGDATA 0x01 /**< data put on overflow page */ #define F_SUBDATA 0x02 /**< data is a sub-database */ #define F_DUPDATA 0x04 /**< data has duplicates */ + +/** valid flags for #mdb_node_add() */ +#define NODE_ADD_FLAGS (F_DUPDATA|F_SUBDATA|MDB_RESERVE|MDB_APPEND) + /** @} */ unsigned short mn_flags; /**< @ref mdb_node */ unsigned short mn_ksize; /**< key size */ @@ -614,6 +697,30 @@ typedef struct MDB_node { /** The size of a key in a node */ #define NODEKSZ(node) ((node)->mn_ksize) + /** Copy a page number from src to dst */ +#ifdef MISALIGNED_OK +#define COPY_PGNO(dst,src) dst = src +#else +#if SIZE_MAX > 4294967295UL +#define COPY_PGNO(dst,src) do { \ + unsigned short *s, *d; \ + s = (unsigned short *)&(src); \ + d = (unsigned short *)&(dst); \ + *d++ = *s++; \ + *d++ = *s++; \ + *d++ = *s++; \ + *d = *s; \ +} while (0) +#else +#define COPY_PGNO(dst,src) do { \ + unsigned short *s, *d; \ + s = (unsigned short *)&(src); \ + d = (unsigned short *)&(dst); \ + *d++ = *s++; \ + *d = *s; \ +} while (0) +#endif +#endif /** The address of a key in a LEAF2 page. * LEAF2 pages are used for #MDB_DUPFIXED sorted-duplicate sub-DBs. * There are no node headers, keys are stored contiguously. @@ -641,12 +748,9 @@ typedef struct MDB_db { /** Handle for the default DB. */ #define MAIN_DBI 1 - /** Identify a data item as a valid sub-DB record */ -#define MDB_SUBDATA 0x8200 - /** Meta page content. */ typedef struct MDB_meta { - /** Stamp identifying this as an MDB data file. It must be set + /** Stamp identifying this as an MDB file. It must be set * to #MDB_MAGIC. */ uint32_t mm_magic; /** Version number of this lock file. Must be set to #MDB_VERSION. */ @@ -662,12 +766,23 @@ typedef struct MDB_meta { txnid_t mm_txnid; /**< txnid that committed this page */ } MDB_meta; + /** Buffer for a stack-allocated dirty page. + * The members define size and alignment, and silence type + * aliasing warnings. They are not used directly; that could + * mean incorrectly using several union members in parallel. + */ +typedef union MDB_pagebuf { + char mb_raw[MDB_PAGESIZE]; + MDB_page mb_page; + struct { + char mm_pad[PAGEHDRSZ]; + MDB_meta mm_meta; + } mb_metabuf; +} MDB_pagebuf; + /** Auxiliary DB info. * The information here is mostly static/read-only. There is * only a single copy of this record in the environment. - * The \b md_dirty flag is not read-only, but only a write - * transaction can ever update it, and only write transactions - * need to worry about it. */ typedef struct MDB_dbx { MDB_val md_name; /**< name of the database */ @@ -675,14 +790,14 @@ typedef struct MDB_dbx { MDB_cmp_func *md_dcmp; /**< function for comparing data items */ MDB_rel_func *md_rel; /**< user relocate function */ void *md_relctx; /**< user-provided context for md_rel */ - MDB_dbi md_parent; /**< parent DB of a sub-DB */ - unsigned int md_dirty; /**< TRUE if DB was written in this txn */ } MDB_dbx; /** A database transaction. * Every operation requires a transaction handle. */ struct MDB_txn { + MDB_txn *mt_parent; /**< parent of a nested txn */ + MDB_txn *mt_child; /**< nested txn under this txn */ 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 @@ -691,9 +806,8 @@ struct MDB_txn { txnid_t mt_txnid; MDB_env *mt_env; /**< the DB environment */ /** The list of pages that became unused during this transaction. - * This is an #IDL. */ - pgno_t *mt_free_pgs; + IDL mt_free_pgs; union { ID2L dirty_list; /**< modified pages */ MDB_reader *reader; /**< this thread's slot in the reader table */ @@ -702,14 +816,30 @@ struct MDB_txn { MDB_dbx *mt_dbxs; /** Array of MDB_db records for each known DB */ MDB_db *mt_dbs; +/** @defgroup mt_dbflag Transaction DB Flags + * @ingroup internal + * @{ + */ +#define DB_DIRTY 0x01 /**< DB was written in this txn */ +#define DB_STALE 0x02 /**< DB record is older than txnID */ +/** @} */ + /** 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. */ MDB_dbi mt_numdbs; +/** @defgroup mdb_txn Transaction Flags + * @ingroup internal + * @{ + */ #define MDB_TXN_RDONLY 0x01 /**< read-only transaction */ #define MDB_TXN_ERROR 0x02 /**< an error has occurred */ - unsigned int mt_flags; +/** @} */ + unsigned int mt_flags; /**< @ref mdb_txn */ /** Tracks which of the two meta pages was used at the start * of this transaction. */ @@ -726,6 +856,10 @@ struct MDB_xcursor; /** Cursors are used for all DB operations */ struct MDB_cursor { + /** Next cursor on this DB in this txn */ + MDB_cursor *mc_next; + /** Original cursor if this is a shadow */ + MDB_cursor *mc_orig; /** Context used for databases with #MDB_DUPSORT, otherwise NULL */ struct MDB_xcursor *mc_xcursor; /** The transaction that owns this cursor */ @@ -736,8 +870,10 @@ struct MDB_cursor { MDB_db *mc_db; /** The database auxiliary record for this cursor */ MDB_dbx *mc_dbx; + /** The @ref mt_dbflag for this database */ + unsigned char *mc_dbflag; unsigned short mc_snum; /**< number of pushed pages */ - unsigned short mc_top; /**< index of top page, mc_snum-1 */ + unsigned short mc_top; /**< index of top page, normally mc_snum-1 */ /** @defgroup mdb_cursor Cursor Flags * @ingroup internal * Cursor state flags. @@ -745,6 +881,9 @@ struct MDB_cursor { */ #define C_INITIALIZED 0x01 /**< cursor has been initialized and is valid */ #define C_EOF 0x02 /**< No more data */ +#define C_SUB 0x04 /**< Cursor is a sub-cursor */ +#define C_SHADOW 0x08 /**< Cursor is a dup from a parent txn */ +#define C_ALLOCD 0x10 /**< Cursor was malloc'd */ /** @} */ unsigned int mc_flags; /**< @ref mdb_cursor */ MDB_page *mc_pg[CURSOR_STACK]; /**< stack of pushed pages */ @@ -763,6 +902,8 @@ typedef struct MDB_xcursor { MDB_db mx_db; /** The auxiliary DB record for this Dup DB */ MDB_dbx mx_dbx; + /** The @ref mt_dbflag for this Dup DB */ + unsigned char mx_dbflag; } MDB_xcursor; /** A set of pages freed by an earlier transaction. */ @@ -799,13 +940,16 @@ struct MDB_env { pgno_t me_maxpg; /**< me_mapsize / me_psize */ unsigned int me_psize; /**< size of a page, from #GET_PAGESIZE */ unsigned int me_db_toggle; /**< which DB table is current */ + txnid_t me_wtxnid; /**< ID of last txn we committed */ + txnid_t me_pgfirst; /**< ID of first old page record we used */ + txnid_t me_pglast; /**< ID of last old page record we used */ MDB_dbx *me_dbxs; /**< array of static DB info */ MDB_db *me_dbs[2]; /**< two arrays of MDB_db info */ MDB_oldpages *me_pghead; /**< list of old page records */ pthread_key_t me_txkey; /**< thread-key for readers */ MDB_page *me_dpages; /**< list of malloc'd blocks for re-use */ /** IDL of pages that became unused in a write txn */ - pgno_t me_free_pgs[MDB_IDL_UM_SIZE]; + IDL me_free_pgs; /** ID2L of pages that were written during a write txn */ ID2 me_dirty_list[MDB_IDL_UM_SIZE]; /** rwlock for the DB tables, if #LAZY_LOCKS is false */ @@ -814,9 +958,17 @@ struct MDB_env { HANDLE me_rmutex; /* Windows mutexes don't reside in shared mem */ HANDLE me_wmutex; #endif +#ifdef __APPLE__ + sem_t *me_rmutex; /* Apple doesn't support shared mutexes */ + sem_t *me_wmutex; +#endif }; /** max number of pages to commit in one writev() call */ #define MDB_COMMIT_PAGES 64 +#if defined(IOV_MAX) && IOV_MAX < MDB_COMMIT_PAGES +#undef MDB_COMMIT_PAGES +#define MDB_COMMIT_PAGES IOV_MAX +#endif static MDB_page *mdb_page_alloc(MDB_cursor *mc, int num); static MDB_page *mdb_page_new(MDB_cursor *mc, uint32_t flags, int num); @@ -829,7 +981,7 @@ static int mdb_page_search(MDB_cursor *mc, MDB_val *key, int modify); static int mdb_page_merge(MDB_cursor *csrc, MDB_cursor *cdst); static int mdb_page_split(MDB_cursor *mc, MDB_val *newkey, MDB_val *newdata, - pgno_t newpgno); + pgno_t newpgno, unsigned int nflags); static int mdb_env_read_header(MDB_env *env, MDB_meta *meta); static int mdb_env_read_meta(MDB_env *env, int *which); @@ -837,8 +989,9 @@ static int mdb_env_write_meta(MDB_txn *txn); static MDB_node *mdb_node_search(MDB_cursor *mc, MDB_val *key, int *exactp); static int mdb_node_add(MDB_cursor *mc, indx_t indx, - MDB_val *key, MDB_val *data, pgno_t pgno, uint8_t flags); + MDB_val *key, MDB_val *data, pgno_t pgno, unsigned int flags); static void mdb_node_del(MDB_page *mp, indx_t indx, int ksize); +static void mdb_node_shrink(MDB_page *mp, indx_t indx); static int mdb_node_move(MDB_cursor *csrc, MDB_cursor *cdst); static int mdb_node_read(MDB_txn *txn, MDB_node *leaf, MDB_val *data); static size_t mdb_leaf_size(MDB_env *env, MDB_val *key, MDB_val *data); @@ -863,6 +1016,7 @@ static void mdb_cursor_init(MDB_cursor *mc, MDB_txn *txn, MDB_dbi dbi, MDB_xcurs static void mdb_xcursor_init0(MDB_cursor *mc); static void mdb_xcursor_init1(MDB_cursor *mc, MDB_node *node); +static int mdb_drop0(MDB_cursor *mc, int subs); static void mdb_default_cmp(MDB_txn *txn, MDB_dbi dbi); /** @cond */ @@ -907,7 +1061,7 @@ mdb_strerror(int err) return strerror(err); } -#if DEBUG +#if MDB_DEBUG /** Display a key in hexadecimal and return the address of the result. * @param[in] key the key to display * @param[in] buf the buffer to write into. Should always be #DKBUF. @@ -925,6 +1079,7 @@ mdb_dkey(MDB_val *key, char *buf) * printable characters, print it as-is instead of converting to hex. */ #if 1 + buf[0] = '\0'; for (i=0; imv_size; i++) ptr += sprintf(ptr, "%02x", *c++); #else @@ -932,6 +1087,74 @@ mdb_dkey(MDB_val *key, char *buf) #endif return buf; } + +/** Display all the keys in the page. */ +static void +mdb_page_keys(MDB_page *mp) +{ + MDB_node *node; + unsigned int i, nkeys; + MDB_val key; + DKBUF; + + nkeys = NUMKEYS(mp); + DPRINTF("numkeys %d", nkeys); + for (i=0; imn_ksize; + key.mv_data = node->mn_data; + DPRINTF("key %d: %s", i, DKEY(&key)); + } +} +#endif + +#if MDB_DEBUG > 2 +/** Count all the pages in each DB and in the freelist + * and make sure it matches the actual number of pages + * being used. + */ +static void mdb_audit(MDB_txn *txn) +{ + MDB_cursor mc; + MDB_val key, data; + int rc, i; + ID freecount, count; + + freecount = 0; + mdb_cursor_init(&mc, txn, FREE_DBI, NULL); + while ((rc = mdb_cursor_get(&mc, &key, &data, MDB_NEXT)) == 0) + freecount += *(ID *)data.mv_data; + freecount += txn->mt_dbs[0].md_branch_pages + txn->mt_dbs[0].md_leaf_pages + + txn->mt_dbs[0].md_overflow_pages; + + count = 0; + for (i = 0; imt_numdbs; i++) { + count += txn->mt_dbs[i].md_branch_pages + + txn->mt_dbs[i].md_leaf_pages + + txn->mt_dbs[i].md_overflow_pages; + if (txn->mt_dbs[i].md_flags & MDB_DUPSORT) { + MDB_xcursor mx; + mdb_cursor_init(&mc, txn, i, &mx); + mdb_page_search(&mc, NULL, 0); + do { + int j; + MDB_page *mp; + mp = mc.mc_pg[mc.mc_top]; + for (j=0; jmn_flags & F_SUBDATA) { + MDB_db db; + memcpy(&db, NODEDATA(leaf), sizeof(db)); + count += db.md_branch_pages + db.md_leaf_pages + + db.md_overflow_pages; + } + } + } + while (mdb_cursor_sibling(&mc, 1) == 0); + } + } + assert(freecount + count + 2 >= txn->mt_next_pgno - 1); +} #endif int @@ -949,6 +1172,23 @@ mdb_dcmp(MDB_txn *txn, MDB_dbi dbi, const MDB_val *a, const MDB_val *b) return EINVAL; /* too bad you can't distinguish this from a valid result */ } +/** Allocate a single page. + * Re-use old malloc'd pages first, otherwise just malloc. + */ +static MDB_page * +mdb_page_malloc(MDB_cursor *mc) { + MDB_page *ret; + size_t sz = mc->mc_txn->mt_env->me_psize; + if ((ret = mc->mc_txn->mt_env->me_dpages) != NULL) { + VGMEMP_ALLOC(mc->mc_txn->mt_env, ret, sz); + VGMEMP_DEFINED(ret, sizeof(ret->mp_next)); + mc->mc_txn->mt_env->me_dpages = ret->mp_next; + } else if ((ret = malloc(sz)) != NULL) { + VGMEMP_ALLOC(mc->mc_txn->mt_env, ret, sz); + } + return ret; +} + /** Allocate pages for writing. * If there are free pages available from older transactions, they * will be re-used first. Otherwise a new page will be allocated. @@ -968,17 +1208,32 @@ mdb_page_alloc(MDB_cursor *mc, int num) if (txn->mt_txnid > 2) { - if (!txn->mt_env->me_pghead && mc->mc_dbi != FREE_DBI && + if (!txn->mt_env->me_pghead && txn->mt_dbs[FREE_DBI].md_root != P_INVALID) { /* See if there's anything in the free DB */ MDB_cursor m2; MDB_node *leaf; - txnid_t *kptr, oldest; + MDB_val data; + txnid_t *kptr, oldest, last; mdb_cursor_init(&m2, txn, FREE_DBI, NULL); - mdb_page_search(&m2, NULL, 0); - leaf = NODEPTR(m2.mc_pg[m2.mc_top], 0); - kptr = (txnid_t *)NODEKEY(leaf); + if (!txn->mt_env->me_pgfirst) { + mdb_page_search(&m2, NULL, 0); + leaf = NODEPTR(m2.mc_pg[m2.mc_top], 0); + kptr = (txnid_t *)NODEKEY(leaf); + last = *kptr; + } else { + MDB_val key; + int rc, exact = 0; + last = txn->mt_env->me_pglast + 1; + leaf = NULL; + key.mv_data = &last; + key.mv_size = sizeof(last); + rc = mdb_cursor_set(&m2, &key, &data, MDB_SET, &exact); + if (rc) + goto none; + last = *(txnid_t *)key.mv_data; + } { unsigned int i; @@ -990,22 +1245,26 @@ mdb_page_alloc(MDB_cursor *mc, int num) } } - if (oldest > *kptr) { + if (oldest > last) { /* It's usable, grab it. */ MDB_oldpages *mop; - MDB_val data; pgno_t *idl; - mdb_node_read(txn, leaf, &data); + if (!txn->mt_env->me_pgfirst) { + mdb_node_read(txn, leaf, &data); + } idl = (ID *) data.mv_data; mop = malloc(sizeof(MDB_oldpages) + MDB_IDL_SIZEOF(idl) - sizeof(pgno_t)); mop->mo_next = txn->mt_env->me_pghead; - mop->mo_txnid = *kptr; + mop->mo_txnid = last; + txn->mt_env->me_pglast = last; + if (!txn->mt_env->me_pgfirst) + txn->mt_env->me_pgfirst = last; txn->mt_env->me_pghead = mop; memcpy(mop->mo_pages, idl, MDB_IDL_SIZEOF(idl)); -#if DEBUG > 1 +#if MDB_DEBUG > 1 { unsigned int i; DPRINTF("IDL read txn %zu root %zu num %zu", @@ -1015,12 +1274,9 @@ mdb_page_alloc(MDB_cursor *mc, int num) } } #endif - /* drop this IDL from the DB */ - m2.mc_ki[m2.mc_top] = 0; - m2.mc_flags = C_INITIALIZED; - mdb_cursor_del(&m2, 0); } } +none: if (txn->mt_env->me_pghead) { MDB_oldpages *mop = txn->mt_env->me_pghead; if (num > 1) { @@ -1050,16 +1306,20 @@ mdb_page_alloc(MDB_cursor *mc, int num) if (pgno == P_INVALID) { /* DB size is maxed out */ if (txn->mt_next_pgno + num >= txn->mt_env->me_maxpg) { - assert(txn->mt_next_pgno + num < txn->mt_env->me_maxpg); + DPUTS("DB size maxed out"); return NULL; } } if (txn->mt_env->me_dpages && num == 1) { np = txn->mt_env->me_dpages; + VGMEMP_ALLOC(txn->mt_env, np, txn->mt_env->me_psize); + VGMEMP_DEFINED(np, sizeof(np->mp_next)); txn->mt_env->me_dpages = np->mp_next; } else { - if ((np = malloc(txn->mt_env->me_psize * num )) == NULL) + size_t sz = txn->mt_env->me_psize * num; + if ((np = malloc(sz)) == NULL) return NULL; + VGMEMP_ALLOC(txn->mt_env, np, sz); } if (pgno == P_INVALID) { np->mp_pgno = txn->mt_next_pgno; @@ -1090,19 +1350,70 @@ mdb_page_touch(MDB_cursor *mc) return ENOMEM; DPRINTF("touched db %u page %zu -> %zu", mc->mc_dbi, mp->mp_pgno, np->mp_pgno); assert(mp->mp_pgno != np->mp_pgno); - mdb_midl_append(mc->mc_txn->mt_free_pgs, mp->mp_pgno); + mdb_midl_append(&mc->mc_txn->mt_free_pgs, mp->mp_pgno); pgno = np->mp_pgno; memcpy(np, mp, mc->mc_txn->mt_env->me_psize); mp = np; mp->mp_pgno = pgno; mp->mp_flags |= P_DIRTY; +finish: + /* Adjust other cursors pointing to mp */ + if (mc->mc_flags & C_SUB) { + MDB_cursor *m2, *m3; + MDB_dbi dbi = mc->mc_dbi-1; + + for (m2 = mc->mc_txn->mt_cursors[dbi]; m2; m2=m2->mc_next) { + if (m2 == mc) continue; + m3 = &m2->mc_xcursor->mx_cursor; + if (m3->mc_snum < mc->mc_snum) continue; + if (m3->mc_pg[mc->mc_top] == mc->mc_pg[mc->mc_top]) { + m3->mc_pg[mc->mc_top] = mp; + } + } + } else { + MDB_cursor *m2; + + for (m2 = mc->mc_txn->mt_cursors[mc->mc_dbi]; m2; m2=m2->mc_next) { + if (m2 == mc || m2->mc_snum < mc->mc_snum) continue; + if (m2->mc_pg[mc->mc_top] == mc->mc_pg[mc->mc_top]) { + m2->mc_pg[mc->mc_top] = mp; + } + } + } mc->mc_pg[mc->mc_top] = mp; /** If this page has a parent, update the parent to point to * this new page. */ if (mc->mc_top) SETPGNO(NODEPTR(mc->mc_pg[mc->mc_top-1], mc->mc_ki[mc->mc_top-1]), mp->mp_pgno); + else + mc->mc_db->md_root = mp->mp_pgno; + } else if (mc->mc_txn->mt_parent) { + MDB_page *np; + ID2 mid; + /* If txn has a parent, make sure the page is in our + * dirty list. + */ + if (mc->mc_txn->mt_u.dirty_list[0].mid) { + unsigned x = mdb_mid2l_search(mc->mc_txn->mt_u.dirty_list, mp->mp_pgno); + if (x <= mc->mc_txn->mt_u.dirty_list[0].mid && + mc->mc_txn->mt_u.dirty_list[x].mid == mp->mp_pgno) { + if (mc->mc_txn->mt_u.dirty_list[x].mptr != mp) { + mp = mc->mc_txn->mt_u.dirty_list[x].mptr; + mc->mc_pg[mc->mc_top] = mp; + } + return 0; + } + } + /* No - copy it */ + np = mdb_page_malloc(mc); + memcpy(np, mp, mc->mc_txn->mt_env->me_psize); + mid.mid = np->mp_pgno; + mid.mptr = np; + mdb_mid2l_insert(mc->mc_txn->mt_u.dirty_list, &mid); + mp = np; + goto finish; } return 0; } @@ -1112,13 +1423,100 @@ mdb_env_sync(MDB_env *env, int force) { int rc = 0; if (force || !F_ISSET(env->me_flags, MDB_NOSYNC)) { - if (fdatasync(env->me_fd)) + if (MDB_FDATASYNC(env->me_fd)) rc = ErrCode(); } return rc; } -static inline void +/** Make shadow copies of all of parent txn's cursors */ +static int +mdb_cursor_shadow(MDB_txn *src, MDB_txn *dst) +{ + MDB_cursor *mc, *m2; + unsigned int i, j, size; + + for (i=0;imt_numdbs; i++) { + if (src->mt_cursors[i]) { + size = sizeof(MDB_cursor); + if (src->mt_cursors[i]->mc_xcursor) + size += sizeof(MDB_xcursor); + for (m2 = src->mt_cursors[i]; m2; m2=m2->mc_next) { + mc = malloc(size); + if (!mc) + return ENOMEM; + mc->mc_orig = m2; + mc->mc_txn = dst; + mc->mc_dbi = i; + mc->mc_db = &dst->mt_dbs[i]; + mc->mc_dbx = m2->mc_dbx; + mc->mc_dbflag = &dst->mt_dbflags[i]; + mc->mc_snum = m2->mc_snum; + mc->mc_top = m2->mc_top; + mc->mc_flags = m2->mc_flags | C_SHADOW; + for (j=0; jmc_snum; j++) { + mc->mc_pg[j] = m2->mc_pg[j]; + mc->mc_ki[j] = m2->mc_ki[j]; + } + if (m2->mc_xcursor) { + MDB_xcursor *mx, *mx2; + mx = (MDB_xcursor *)(mc+1); + mc->mc_xcursor = mx; + mx2 = m2->mc_xcursor; + mx->mx_db = mx2->mx_db; + mx->mx_dbx = mx2->mx_dbx; + mx->mx_dbflag = mx2->mx_dbflag; + mx->mx_cursor.mc_txn = dst; + mx->mx_cursor.mc_dbi = mx2->mx_cursor.mc_dbi; + mx->mx_cursor.mc_db = &mx->mx_db; + mx->mx_cursor.mc_dbx = &mx->mx_dbx; + mx->mx_cursor.mc_dbflag = &mx->mx_dbflag; + mx->mx_cursor.mc_snum = mx2->mx_cursor.mc_snum; + mx->mx_cursor.mc_top = mx2->mx_cursor.mc_top; + mx->mx_cursor.mc_flags = mx2->mx_cursor.mc_flags | C_SHADOW; + for (j=0; jmx_cursor.mc_snum; j++) { + mx->mx_cursor.mc_pg[j] = mx2->mx_cursor.mc_pg[j]; + mx->mx_cursor.mc_ki[j] = mx2->mx_cursor.mc_ki[j]; + } + } else { + mc->mc_xcursor = NULL; + } + mc->mc_next = dst->mt_cursors[i]; + dst->mt_cursors[i] = mc; + } + } + } + return MDB_SUCCESS; +} + +/** Merge shadow cursors back into parent's */ +static void +mdb_cursor_merge(MDB_txn *txn) +{ + MDB_dbi i; + for (i=0; imt_numdbs; i++) { + if (txn->mt_cursors[i]) { + MDB_cursor *mc; + while ((mc = txn->mt_cursors[i])) { + txn->mt_cursors[i] = mc->mc_next; + if (mc->mc_flags & C_SHADOW) { + MDB_cursor *m2 = mc->mc_orig; + unsigned int j; + m2->mc_snum = mc->mc_snum; + m2->mc_top = mc->mc_top; + for (j=0; jmc_snum; j++) { + m2->mc_pg[j] = mc->mc_pg[j]; + m2->mc_ki[j] = mc->mc_ki[j]; + } + } + if (mc->mc_flags & C_ALLOCD) + free(mc); + } + } + } +} + +static void mdb_txn_reset0(MDB_txn *txn); /** Common code for #mdb_txn_begin() and #mdb_txn_renew(). @@ -1127,10 +1525,11 @@ mdb_txn_reset0(MDB_txn *txn); * fail for read-only transactions, and then only if the * reader table is full. */ -static inline int +static int mdb_txn_renew0(MDB_txn *txn) { MDB_env *env = txn->mt_env; + char mt_dbflag = 0; if (txn->mt_flags & MDB_TXN_RDONLY) { MDB_reader *r = pthread_getspecific(env->me_txkey); @@ -1155,14 +1554,22 @@ mdb_txn_renew0(MDB_txn *txn) r = &env->me_txns->mti_readers[i]; pthread_setspecific(env->me_txkey, r); } - txn->mt_txnid = env->me_txns->mti_txnid; txn->mt_toggle = env->me_txns->mti_me_toggle; + txn->mt_txnid = env->me_txns->mti_txnid; + /* This happens if a different process was the + * last writer to the DB. + */ + if (env->me_wtxnid < txn->mt_txnid) + mt_dbflag = DB_STALE; r->mr_txnid = txn->mt_txnid; txn->mt_u.reader = r; } else { LOCK_MUTEX_W(env); - txn->mt_txnid = env->me_txns->mti_txnid+1; + txn->mt_txnid = env->me_txns->mti_txnid; + if (env->me_wtxnid < txn->mt_txnid) + mt_dbflag = DB_STALE; + txn->mt_txnid++; txn->mt_toggle = env->me_txns->mti_me_toggle; txn->mt_u.dirty_list = env->me_dirty_list; txn->mt_u.dirty_list[0].mid = 0; @@ -1182,6 +1589,8 @@ mdb_txn_renew0(MDB_txn *txn) (txn->mt_numdbs - 2) * sizeof(MDB_db)); LAZY_RWLOCK_UNLOCK(&env->me_dblock); + memset(txn->mt_dbflags, mt_dbflag, env->me_numdbs); + return MDB_SUCCESS; } @@ -1208,26 +1617,67 @@ mdb_txn_renew(MDB_txn *txn) } int -mdb_txn_begin(MDB_env *env, unsigned int flags, MDB_txn **ret) +mdb_txn_begin(MDB_env *env, MDB_txn *parent, unsigned int flags, MDB_txn **ret) { MDB_txn *txn; - int rc; + int rc, size; if (env->me_flags & MDB_FATAL_ERROR) { DPUTS("environment had fatal error, must shutdown!"); return MDB_PANIC; } - if ((txn = calloc(1, sizeof(MDB_txn) + env->me_maxdbs * sizeof(MDB_db))) == NULL) { + if (parent) { + /* parent already has an active child txn */ + if (parent->mt_child) { + return EINVAL; + } + } + size = sizeof(MDB_txn) + 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())); return ENOMEM; } txn->mt_dbs = (MDB_db *)(txn+1); 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_env = env; - rc = mdb_txn_renew0(txn); + if (parent) { + txn->mt_free_pgs = mdb_midl_alloc(); + if (!txn->mt_free_pgs) { + free(txn); + return ENOMEM; + } + txn->mt_u.dirty_list = malloc(sizeof(ID2)*MDB_IDL_UM_SIZE); + if (!txn->mt_u.dirty_list) { + free(txn->mt_free_pgs); + free(txn); + return ENOMEM; + } + txn->mt_txnid = parent->mt_txnid; + txn->mt_toggle = parent->mt_toggle; + txn->mt_u.dirty_list[0].mid = 0; + txn->mt_free_pgs[0] = 0; + txn->mt_next_pgno = parent->mt_next_pgno; + parent->mt_child = txn; + txn->mt_parent = parent; + txn->mt_numdbs = parent->mt_numdbs; + txn->mt_dbxs = parent->mt_dbxs; + memcpy(txn->mt_dbs, parent->mt_dbs, txn->mt_numdbs * sizeof(MDB_db)); + memcpy(txn->mt_dbflags, parent->mt_dbflags, txn->mt_numdbs); + mdb_cursor_shadow(parent, txn); + rc = 0; + } else { + rc = mdb_txn_renew0(txn); + } if (rc) free(txn); else { @@ -1243,7 +1693,7 @@ mdb_txn_begin(MDB_env *env, unsigned int flags, MDB_txn **ret) /** Common code for #mdb_txn_reset() and #mdb_txn_abort(). * @param[in] txn the transaction handle to reset */ -static inline void +static void mdb_txn_reset0(MDB_txn *txn) { MDB_env *env = txn->mt_env; @@ -1253,29 +1703,52 @@ mdb_txn_reset0(MDB_txn *txn) } else { MDB_oldpages *mop; MDB_page *dp; - MDB_dbi dbi; unsigned int i; + /* close(free) all cursors */ + for (i=0; imt_numdbs; i++) { + if (txn->mt_cursors[i]) { + MDB_cursor *mc; + while ((mc = txn->mt_cursors[i])) { + txn->mt_cursors[i] = mc->mc_next; + if (mc->mc_flags & C_ALLOCD) + free(mc); + } + } + } + /* return all dirty pages to dpage list */ for (i=1; i<=txn->mt_u.dirty_list[0].mid; i++) { dp = txn->mt_u.dirty_list[i].mptr; if (!IS_OVERFLOW(dp) || dp->mp_pages == 1) { dp->mp_next = txn->mt_env->me_dpages; + VGMEMP_FREE(txn->mt_env, dp); txn->mt_env->me_dpages = dp; } else { /* large pages just get freed directly */ + VGMEMP_FREE(txn->mt_env, dp); free(dp); } } + if (txn->mt_parent) { + txn->mt_parent->mt_child = NULL; + free(txn->mt_free_pgs); + free(txn->mt_u.dirty_list); + return; + } else { + if (mdb_midl_shrink(&txn->mt_free_pgs)) + env->me_free_pgs = txn->mt_free_pgs; + } + while ((mop = txn->mt_env->me_pghead)) { txn->mt_env->me_pghead = mop->mo_next; free(mop); } + txn->mt_env->me_pgfirst = 0; + txn->mt_env->me_pglast = 0; env->me_txn = NULL; - for (dbi=2; dbime_numdbs; dbi++) - env->me_dbxs[dbi].md_dirty = 0; /* The writer mutex was locked in mdb_txn_begin. */ UNLOCK_MUTEX_W(env); } @@ -1304,6 +1777,9 @@ mdb_txn_abort(MDB_txn *txn) txn->mt_txnid, (txn->mt_flags & MDB_TXN_RDONLY) ? 'r' : 'w', (void *)txn, (void *)txn->mt_env, txn->mt_dbs[MAIN_DBI].md_root); + if (txn->mt_child) + mdb_txn_abort(txn->mt_child); + mdb_txn_reset0(txn); free(txn); } @@ -1317,27 +1793,101 @@ mdb_txn_commit(MDB_txn *txn) off_t size; MDB_page *dp; MDB_env *env; - pgno_t next; + pgno_t next, freecnt; MDB_cursor mc; assert(txn != NULL); assert(txn->mt_env != NULL); + if (txn->mt_child) { + mdb_txn_commit(txn->mt_child); + txn->mt_child = NULL; + } + env = txn->mt_env; if (F_ISSET(txn->mt_flags, MDB_TXN_RDONLY)) { + if (txn->mt_numdbs > env->me_numdbs) { + /* update the DB tables */ + int toggle = !env->me_db_toggle; + MDB_db *ip, *jp; + MDB_dbi i; + + ip = &env->me_dbs[toggle][env->me_numdbs]; + jp = &txn->mt_dbs[env->me_numdbs]; + LAZY_RWLOCK_WRLOCK(&env->me_dblock); + for (i = env->me_numdbs; i < txn->mt_numdbs; i++) { + *ip++ = *jp++; + } + + env->me_db_toggle = toggle; + env->me_numdbs = txn->mt_numdbs; + LAZY_RWLOCK_UNLOCK(&env->me_dblock); + } mdb_txn_abort(txn); return MDB_SUCCESS; } - if (txn != env->me_txn) { - DPUTS("attempt to commit unknown transaction"); + if (F_ISSET(txn->mt_flags, MDB_TXN_ERROR)) { + DPUTS("error flag is set, can't commit"); + if (txn->mt_parent) + txn->mt_parent->mt_flags |= MDB_TXN_ERROR; mdb_txn_abort(txn); return EINVAL; } - if (F_ISSET(txn->mt_flags, MDB_TXN_ERROR)) { - DPUTS("error flag is set, can't commit"); + /* Merge (and close) our cursors with parent's */ + mdb_cursor_merge(txn); + + if (txn->mt_parent) { + MDB_db *ip, *jp; + MDB_dbi i; + unsigned x, y; + ID2L dst, src; + + /* Update parent's DB table */ + ip = &txn->mt_parent->mt_dbs[2]; + jp = &txn->mt_dbs[2]; + for (i = 2; i < txn->mt_numdbs; i++) { + if (ip->md_root != jp->md_root) + *ip = *jp; + ip++; jp++; + } + txn->mt_parent->mt_numdbs = txn->mt_numdbs; + + /* Append our free list to parent's */ + mdb_midl_append_list(&txn->mt_parent->mt_free_pgs, + txn->mt_free_pgs); + mdb_midl_free(txn->mt_free_pgs); + + /* Merge our dirty list with parent's */ + dst = txn->mt_parent->mt_u.dirty_list; + src = txn->mt_u.dirty_list; + x = mdb_mid2l_search(dst, src[1].mid); + for (y=1; y<=src[0].mid; y++) { + while (x <= dst[0].mid && dst[x].mid != src[y].mid) x++; + if (x > dst[0].mid) + break; + free(dst[x].mptr); + dst[x].mptr = src[y].mptr; + } + x = dst[0].mid; + for (; y<=src[0].mid; y++) { + if (++x >= MDB_IDL_UM_MAX) { + mdb_txn_abort(txn); + return ENOMEM; + } + dst[x] = src[y]; + } + dst[0].mid = x; + free(txn->mt_u.dirty_list); + txn->mt_parent->mt_child = NULL; + free(txn); + return MDB_SUCCESS; + } + + if (txn != env->me_txn) { + DPUTS("attempt to commit unknown transaction"); mdb_txn_abort(txn); return EINVAL; } @@ -1355,10 +1905,28 @@ mdb_txn_commit(MDB_txn *txn) /* make sure first page of freeDB is touched and on freelist */ mdb_page_search(&mc, NULL, 1); } + + /* Delete IDLs we used from the free list */ + if (env->me_pgfirst) { + txnid_t cur; + MDB_val key; + int exact = 0; + + key.mv_size = sizeof(cur); + for (cur = env->me_pgfirst; cur <= env->me_pglast; cur++) { + key.mv_data = &cur; + + mdb_cursor_set(&mc, &key, NULL, MDB_SET, &exact); + mdb_cursor_del(&mc, 0); + } + env->me_pgfirst = 0; + env->me_pglast = 0; + } + /* save to free list */ + freecnt = txn->mt_free_pgs[0]; if (!MDB_IDL_IS_ZERO(txn->mt_free_pgs)) { MDB_val key, data; - pgno_t i; /* make sure last page of freeDB is touched and on freelist */ key.mv_size = MAXKEYSIZE+1; @@ -1366,7 +1934,7 @@ mdb_txn_commit(MDB_txn *txn) mdb_page_search(&mc, &key, 1); mdb_midl_sort(txn->mt_free_pgs); -#if DEBUG > 1 +#if MDB_DEBUG > 1 { unsigned int i; ID *idl = txn->mt_free_pgs; @@ -1386,28 +1954,51 @@ mdb_txn_commit(MDB_txn *txn) * and make sure the entire thing got written. */ do { - i = txn->mt_free_pgs[0]; + freecnt = txn->mt_free_pgs[0]; data.mv_size = MDB_IDL_SIZEOF(txn->mt_free_pgs); rc = mdb_cursor_put(&mc, &key, &data, 0); if (rc) { mdb_txn_abort(txn); return rc; } - } while (i != txn->mt_free_pgs[0]); + } while (freecnt != txn->mt_free_pgs[0]); + if (mdb_midl_shrink(&txn->mt_free_pgs)) + env->me_free_pgs = txn->mt_free_pgs; } /* should only be one record now */ +again: if (env->me_pghead) { MDB_val key, data; MDB_oldpages *mop; + pgno_t orig; + txnid_t id; mop = env->me_pghead; - key.mv_size = sizeof(pgno_t); - key.mv_data = &mop->mo_txnid; + id = mop->mo_txnid; + key.mv_size = sizeof(id); + key.mv_data = &id; data.mv_size = MDB_IDL_SIZEOF(mop->mo_pages); data.mv_data = mop->mo_pages; + orig = mop->mo_pages[0]; mdb_cursor_put(&mc, &key, &data, 0); - free(env->me_pghead); - env->me_pghead = NULL; + if (mop == env->me_pghead) { + /* could have been used again here */ + if (mop->mo_pages[0] != orig) { + data.mv_size = MDB_IDL_SIZEOF(mop->mo_pages); + data.mv_data = mop->mo_pages; + id = mop->mo_txnid; + mdb_cursor_put(&mc, &key, &data, 0); + } + env->me_pghead = NULL; + free(mop); + } else { + /* was completely used up */ + mdb_cursor_del(&mc, 0); + if (env->me_pghead) + goto again; + } + env->me_pgfirst = 0; + env->me_pglast = 0; } /* Update DB root pointers. Their pages have already been @@ -1420,12 +2011,15 @@ mdb_txn_commit(MDB_txn *txn) mdb_cursor_init(&mc, txn, MAIN_DBI, NULL); for (i = 2; i < txn->mt_numdbs; i++) { - if (txn->mt_dbxs[i].md_dirty) { + if (txn->mt_dbflags[i] & DB_DIRTY) { data.mv_data = &txn->mt_dbs[i]; mdb_cursor_put(&mc, &txn->mt_dbxs[i].md_name, &data, 0); } } } +#if MDB_DEBUG > 2 + mdb_audit(txn); +#endif /* Commit up to MDB_COMMIT_PAGES dirty pages to disk until done. */ @@ -1472,7 +2066,6 @@ mdb_txn_commit(MDB_txn *txn) dp = txn->mt_u.dirty_list[i].mptr; if (dp->mp_pgno != next) { if (n) { - DPRINTF("committing %u dirty pages", n); rc = writev(env->me_fd, iov, n); if (rc != size) { n = ErrCode(); @@ -1492,7 +2085,7 @@ mdb_txn_commit(MDB_txn *txn) DPRINTF("committing page %zu", dp->mp_pgno); iov[n].iov_len = env->me_psize; if (IS_OVERFLOW(dp)) iov[n].iov_len *= dp->mp_pages; - iov[n].iov_base = dp; + iov[n].iov_base = (char *)dp; size += iov[n].iov_len; next = dp->mp_pgno + (IS_OVERFLOW(dp) ? dp->mp_pages : 1); /* clear dirty flag */ @@ -1507,7 +2100,6 @@ mdb_txn_commit(MDB_txn *txn) if (n == 0) break; - DPRINTF("committing %u dirty pages", n); rc = writev(env->me_fd, iov, n); if (rc != size) { n = ErrCode(); @@ -1527,8 +2119,10 @@ mdb_txn_commit(MDB_txn *txn) dp = txn->mt_u.dirty_list[i].mptr; if (!IS_OVERFLOW(dp) || dp->mp_pages == 1) { dp->mp_next = txn->mt_env->me_dpages; + VGMEMP_FREE(txn->mt_env, dp); txn->mt_env->me_dpages = dp; } else { + VGMEMP_FREE(txn->mt_env, dp); free(dp); } txn->mt_u.dirty_list[i].mid = 0; @@ -1540,6 +2134,7 @@ mdb_txn_commit(MDB_txn *txn) mdb_txn_abort(txn); return n; } + env->me_wtxnid = txn->mt_txnid; done: env->me_txn = NULL; @@ -1558,10 +2153,6 @@ done: ip++; jp++; } - for (i = 2; i < txn->mt_numdbs; i++) { - if (txn->mt_dbxs[i].md_dirty) - txn->mt_dbxs[i].md_dirty = 0; - } env->me_db_toggle = toggle; env->me_numdbs = txn->mt_numdbs; LAZY_RWLOCK_UNLOCK(&env->me_dblock); @@ -1582,7 +2173,7 @@ done: static int mdb_env_read_header(MDB_env *env, MDB_meta *meta) { - char page[PAGESIZE]; + MDB_pagebuf pbuf; MDB_page *p; MDB_meta *m; int rc, err; @@ -1591,14 +2182,14 @@ mdb_env_read_header(MDB_env *env, MDB_meta *meta) */ #ifdef _WIN32 - if (!ReadFile(env->me_fd, page, PAGESIZE, (DWORD *)&rc, NULL) || rc == 0) + if (!ReadFile(env->me_fd, &pbuf, MDB_PAGESIZE, (DWORD *)&rc, NULL) || rc == 0) #else - if ((rc = read(env->me_fd, page, PAGESIZE)) == 0) + if ((rc = read(env->me_fd, &pbuf, MDB_PAGESIZE)) == 0) #endif { return ENOENT; } - else if (rc != PAGESIZE) { + else if (rc != MDB_PAGESIZE) { err = ErrCode(); if (rc > 0) err = EINVAL; @@ -1606,7 +2197,7 @@ mdb_env_read_header(MDB_env *env, MDB_meta *meta) return err; } - p = (MDB_page *)page; + p = (MDB_page *)&pbuf; if (!F_ISSET(p->mp_flags, P_META)) { DPRINTF("page %zu not a meta page", p->mp_pgno); @@ -1798,11 +2389,17 @@ mdb_env_create(MDB_env **env) if (!e) return ENOMEM; + e->me_free_pgs = mdb_midl_alloc(); + if (!e->me_free_pgs) { + free(e); + return ENOMEM; + } e->me_maxreaders = DEFAULT_READERS; e->me_maxdbs = 2; e->me_fd = INVALID_HANDLE_VALUE; e->me_lfd = INVALID_HANDLE_VALUE; e->me_mfd = INVALID_HANDLE_VALUE; + VGMEMP_CREATE(e,0,0); *env = e; return MDB_SUCCESS; } @@ -1813,6 +2410,8 @@ mdb_env_set_mapsize(MDB_env *env, size_t size) if (env->me_map) return EINVAL; env->me_mapsize = size; + if (env->me_psize) + env->me_maxpg = env->me_mapsize / env->me_psize; return MDB_SUCCESS; } @@ -1899,8 +2498,10 @@ mdb_env_open2(MDB_env *env, unsigned int flags) i |= MAP_FIXED; env->me_map = mmap(meta.mm_address, env->me_mapsize, PROT_READ, i, env->me_fd, 0); - if (env->me_map == MAP_FAILED) + if (env->me_map == MAP_FAILED) { + env->me_map = NULL; return ErrCode(); + } #endif if (newenv) { @@ -1988,6 +2589,80 @@ mdb_env_share_locks(MDB_env *env) } #endif } +#if defined(_WIN32) || defined(__APPLE__) +/* + * hash_64 - 64 bit Fowler/Noll/Vo-0 FNV-1a hash code + * + * @(#) $Revision: 5.1 $ + * @(#) $Id: hash_64a.c,v 5.1 2009/06/30 09:01:38 chongo Exp $ + * @(#) $Source: /usr/local/src/cmd/fnv/RCS/hash_64a.c,v $ + * + * http://www.isthe.com/chongo/tech/comp/fnv/index.html + * + *** + * + * Please do not copyright this code. This code is in the public domain. + * + * LANDON CURT NOLL DISCLAIMS ALL WARRANTIES WITH REGARD TO THIS SOFTWARE, + * INCLUDING ALL IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS. IN NO + * EVENT SHALL LANDON CURT NOLL BE LIABLE FOR ANY SPECIAL, INDIRECT OR + * CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM LOSS OF + * USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR + * OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR + * PERFORMANCE OF THIS SOFTWARE. + * + * By: + * chongo /\oo/\ + * http://www.isthe.com/chongo/ + * + * Share and Enjoy! :-) + */ + +typedef unsigned long long mdb_hash_t; +#define MDB_HASH_INIT ((mdb_hash_t)0xcbf29ce484222325ULL) + +/** perform a 64 bit Fowler/Noll/Vo FNV-1a hash on a buffer + * @param[in] str string to hash + * @param[in] hval initial value for hash + * @return 64 bit hash + * + * NOTE: To use the recommended 64 bit FNV-1a hash, use MDB_HASH_INIT as the + * hval arg on the first call. + */ +static mdb_hash_t +mdb_hash_str(char *str, mdb_hash_t hval) +{ + unsigned char *s = (unsigned char *)str; /* unsigned string */ + /* + * FNV-1a hash each octet of the string + */ + while (*s) { + /* xor the bottom with the current octet */ + hval ^= (mdb_hash_t)*s++; + + /* multiply by the 64 bit FNV magic prime mod 2^64 */ + hval += (hval << 1) + (hval << 4) + (hval << 5) + + (hval << 7) + (hval << 8) + (hval << 40); + } + /* return our new hash value */ + return hval; +} + +/** Hash the string and output the hash in hex. + * @param[in] str string to hash + * @param[out] hexbuf an array of 17 chars to hold the hash + */ +static void +mdb_hash_hex(char *str, char *hexbuf) +{ + int i; + mdb_hash_t h = mdb_hash_str(str, MDB_HASH_INIT); + for (i=0; i<8; i++) { + hexbuf += sprintf(hexbuf, "%02x", (unsigned int)h & 0xff); + h >>= 8; + } +} +#endif /** Open and/or initialize the lock region for the environment. * @param[in] env The MDB environment. @@ -2075,8 +2750,8 @@ mdb_env_setup_locks(MDB_env *env, char *lpath, int mode, int *excl) size = rsize - sizeof(MDB_txninfo); env->me_maxreaders = size/sizeof(MDB_reader) + 1; } -#ifdef _WIN32 { +#ifdef _WIN32 HANDLE mh; mh = CreateFileMapping(env->me_lfd, NULL, PAGE_READWRITE, 0, 0, NULL); @@ -2090,18 +2765,20 @@ mdb_env_setup_locks(MDB_env *env, char *lpath, int mode, int *excl) rc = ErrCode(); goto fail; } - } #else - env->me_txns = mmap(0, rsize, PROT_READ|PROT_WRITE, MAP_SHARED, - env->me_lfd, 0); - if (env->me_txns == MAP_FAILED) { - rc = ErrCode(); - goto fail; - } + void *m = mmap(NULL, rsize, PROT_READ|PROT_WRITE, MAP_SHARED, + env->me_lfd, 0); + if (m == MAP_FAILED) { + env->me_txns = NULL; + rc = ErrCode(); + goto fail; + } + env->me_txns = m; #endif + } if (*excl) { #ifdef _WIN32 - char *ptr; + char hexbuf[17]; if (!mdb_sec_inited) { InitializeSecurityDescriptor(&mdb_null_sd, SECURITY_DESCRIPTOR_REVISION); @@ -2111,28 +2788,46 @@ mdb_env_setup_locks(MDB_env *env, char *lpath, int mode, int *excl) mdb_all_sa.lpSecurityDescriptor = &mdb_null_sd; mdb_sec_inited = 1; } - /* FIXME: only using up to 20 characters of the env path here, - * probably not enough to assure uniqueness... - */ - sprintf(env->me_txns->mti_rmname, "Global\\MDBr%.20s", lpath); - ptr = env->me_txns->mti_rmname + sizeof("Global\\MDBr"); - while ((ptr = strchr(ptr, '\\'))) - *ptr++ = '/'; + mdb_hash_hex(lpath, hexbuf); + sprintf(env->me_txns->mti_rmname, "Global\\MDBr%s", hexbuf); env->me_rmutex = CreateMutex(&mdb_all_sa, FALSE, env->me_txns->mti_rmname); if (!env->me_rmutex) { rc = ErrCode(); goto fail; } - sprintf(env->me_txns->mti_rmname, "Global\\MDBw%.20s", lpath); - ptr = env->me_txns->mti_rmname + sizeof("Global\\MDBw"); - while ((ptr = strchr(ptr, '\\'))) - *ptr++ = '/'; - env->me_wmutex = CreateMutex(&mdb_all_sa, FALSE, env->me_txns->mti_rmname); + sprintf(env->me_txns->mti_wmname, "Global\\MDBw%s", hexbuf); + env->me_wmutex = CreateMutex(&mdb_all_sa, FALSE, env->me_txns->mti_wmname); if (!env->me_wmutex) { rc = ErrCode(); goto fail; } -#else +#else /* _WIN32 */ +#ifdef __APPLE__ + char hexbuf[17]; + mdb_hash_hex(lpath, hexbuf); + sprintf(env->me_txns->mti_rmname, "MDBr%s", hexbuf); + if (sem_unlink(env->me_txns->mti_rmname)) { + rc = ErrCode(); + if (rc != ENOENT && rc != EINVAL) + goto fail; + } + env->me_rmutex = sem_open(env->me_txns->mti_rmname, O_CREAT, mode, 1); + if (!env->me_rmutex) { + rc = ErrCode(); + goto fail; + } + sprintf(env->me_txns->mti_wmname, "MDBw%s", hexbuf); + if (sem_unlink(env->me_txns->mti_wmname)) { + rc = ErrCode(); + if (rc != ENOENT && rc != EINVAL) + goto fail; + } + env->me_wmutex = sem_open(env->me_txns->mti_wmname, O_CREAT, mode, 1); + if (!env->me_wmutex) { + rc = ErrCode(); + goto fail; + } +#else /* __APPLE__ */ pthread_mutexattr_t mattr; pthread_mutexattr_init(&mattr); @@ -2142,7 +2837,8 @@ mdb_env_setup_locks(MDB_env *env, char *lpath, int mode, int *excl) } pthread_mutex_init(&env->me_txns->mti_mutex, &mattr); pthread_mutex_init(&env->me_txns->mti_wmutex, &mattr); -#endif +#endif /* __APPLE__ */ +#endif /* _WIN32 */ env->me_txns->mti_version = MDB_VERSION; env->me_txns->mti_magic = MDB_MAGIC; env->me_txns->mti_txnid = 0; @@ -2176,6 +2872,18 @@ mdb_env_setup_locks(MDB_env *env, char *lpath, int mode, int *excl) rc = ErrCode(); goto fail; } +#endif +#ifdef __APPLE__ + env->me_rmutex = sem_open(env->me_txns->mti_rmname, 0); + if (!env->me_rmutex) { + rc = ErrCode(); + goto fail; + } + env->me_wmutex = sem_open(env->me_txns->mti_wmname, 0); + if (!env->me_wmutex) { + rc = ErrCode(); + goto fail; + } #endif } return MDB_SUCCESS; @@ -2191,6 +2899,9 @@ fail: #define LOCKNAME "/lock.mdb" /** The name of the data file in the DB environment */ #define DATANAME "/data.mdb" + /** The suffix of the lock file when no subdir is used */ +#define LOCKSUFF "-lock" + int mdb_env_open(MDB_env *env, const char *path, unsigned int flags, mode_t mode) { @@ -2198,13 +2909,24 @@ mdb_env_open(MDB_env *env, const char *path, unsigned int flags, mode_t mode) char *lpath, *dpath; len = strlen(path); - lpath = malloc(len + sizeof(LOCKNAME) + len + sizeof(DATANAME)); + if (flags & MDB_NOSUBDIR) { + rc = len + sizeof(LOCKSUFF) + len + 1; + } else { + rc = len + sizeof(LOCKNAME) + len + sizeof(DATANAME); + } + lpath = malloc(rc); if (!lpath) return ENOMEM; - dpath = lpath + len + sizeof(LOCKNAME); - sprintf(lpath, "%s" LOCKNAME, path); - sprintf(dpath, "%s" DATANAME, path); - + if (flags & MDB_NOSUBDIR) { + dpath = lpath + len + sizeof(LOCKSUFF); + sprintf(lpath, "%s" LOCKSUFF, path); + strcpy(dpath, path); + } else { + dpath = lpath + len + sizeof(LOCKNAME); + sprintf(lpath, "%s" LOCKNAME, path); + sprintf(dpath, "%s" DATANAME, path); + } + rc = mdb_env_setup_locks(env, lpath, mode, &excl); if (rc) goto leave; @@ -2288,8 +3010,10 @@ mdb_env_close(MDB_env *env) if (env == NULL) return; + VGMEMP_DESTROY(env); while (env->me_dpages) { dp = env->me_dpages; + VGMEMP_DEFINED(&dp->mp_next, sizeof(dp->mp_next)); env->me_dpages = dp->mp_next; free(dp); } @@ -2313,9 +3037,10 @@ mdb_env_close(MDB_env *env) for (i=0; ime_txns->mti_numreaders; i++) if (env->me_txns->mti_readers[i].mr_pid == pid) env->me_txns->mti_readers[i].mr_pid = 0; - munmap(env->me_txns, (env->me_maxreaders-1)*sizeof(MDB_reader)+sizeof(MDB_txninfo)); + munmap((void *)env->me_txns, (env->me_maxreaders-1)*sizeof(MDB_reader)+sizeof(MDB_txninfo)); } close(env->me_lfd); + mdb_midl_free(env->me_free_pgs); free(env); } @@ -2341,7 +3066,7 @@ mdb_cmp_int(const MDB_val *a, const MDB_val *b) static int mdb_cmp_cint(const MDB_val *a, const MDB_val *b) { -#if __BYTE_ORDER == __LITTLE_ENDIAN +#if BYTE_ORDER == LITTLE_ENDIAN unsigned short *u, *c; int x; @@ -2422,9 +3147,15 @@ mdb_node_search(MDB_cursor *mc, MDB_val *key, int *exactp) nkeys = NUMKEYS(mp); - DPRINTF("searching %u keys in %s page %zu", - nkeys, IS_LEAF(mp) ? "leaf" : "branch", - mp->mp_pgno); +#if MDB_DEBUG + { + pgno_t pgno; + COPY_PGNO(pgno, mp->mp_pgno); + DPRINTF("searching %u keys in %s %spage %zu", + nkeys, IS_LEAF(mp) ? "leaf" : "branch", IS_SUBP(mp) ? "sub-" : "", + pgno); + } +#endif assert(nkeys > 0); @@ -2467,7 +3198,7 @@ mdb_node_search(MDB_cursor *mc, MDB_val *key, int *exactp) nodekey.mv_data = NODEKEY(node); rc = cmp(key, &nodekey); -#if DEBUG +#if MDB_DEBUG if (IS_LEAF(mp)) DPRINTF("found leaf index %u [%s], rc = %i", i, DKEY(&nodekey), rc); @@ -2501,6 +3232,20 @@ mdb_node_search(MDB_cursor *mc, MDB_val *key, int *exactp) return node; } +#if 0 +static void +mdb_cursor_adjust(MDB_cursor *mc, func) +{ + MDB_cursor *m2; + + for (m2 = mc->mc_txn->mt_cursors[mc->mc_dbi]; m2; m2=m2->mc_next) { + if (m2->mc_pg[m2->mc_top] == mc->mc_pg[mc->mc_top]) { + func(mc, m2); + } + } +} +#endif + /** Pop a page off the top of the cursor's stack. */ static void mdb_cursor_pop(MDB_cursor *mc) @@ -2667,14 +3412,41 @@ mdb_page_search(MDB_cursor *mc, MDB_val *key, int modify) if (F_ISSET(mc->mc_txn->mt_flags, MDB_TXN_ERROR)) { DPUTS("transaction has failed, must abort"); return EINVAL; - } else + } else { + /* Make sure we're using an up-to-date root */ + if (mc->mc_dbi > MAIN_DBI) { + if ((*mc->mc_dbflag & DB_STALE) || + (modify && !(*mc->mc_dbflag & DB_DIRTY))) { + MDB_cursor mc2; + unsigned char dbflag = 0; + mdb_cursor_init(&mc2, mc->mc_txn, MAIN_DBI, NULL); + rc = mdb_page_search(&mc2, &mc->mc_dbx->md_name, modify); + if (rc) + return rc; + if (*mc->mc_dbflag & DB_STALE) { + MDB_val data; + int exact = 0; + MDB_node *leaf = mdb_node_search(&mc2, + &mc->mc_dbx->md_name, &exact); + if (!exact) + return MDB_NOTFOUND; + mdb_node_read(mc->mc_txn, leaf, &data); + memcpy(mc->mc_db, data.mv_data, sizeof(MDB_db)); + } + if (modify) + dbflag = DB_DIRTY; + *mc->mc_dbflag = dbflag; + } + } root = mc->mc_db->md_root; - if (root == P_INVALID) { /* Tree is empty. */ - DPUTS("tree is empty"); - return MDB_NOTFOUND; + if (root == P_INVALID) { /* Tree is empty. */ + DPUTS("tree is empty"); + return MDB_NOTFOUND; + } } + assert(root > 1); if ((rc = mdb_page_get(mc->mc_txn, root, &mc->mc_pg[0]))) return rc; @@ -2685,20 +3457,8 @@ mdb_page_search(MDB_cursor *mc, MDB_val *key, int modify) mc->mc_dbi, root, mc->mc_pg[0]->mp_flags); if (modify) { - /* For sub-databases, update main root first */ - if (mc->mc_dbi > MAIN_DBI && !mc->mc_dbx->md_dirty) { - MDB_cursor mc2; - mdb_cursor_init(&mc2, mc->mc_txn, MAIN_DBI, NULL); - rc = mdb_page_search(&mc2, &mc->mc_dbx->md_name, 1); - if (rc) - return rc; - mc->mc_dbx->md_dirty = 1; - } - if (!F_ISSET(mc->mc_pg[0]->mp_flags, P_DIRTY)) { - if ((rc = mdb_page_touch(mc))) - return rc; - mc->mc_db->md_root = mc->mc_pg[0]->mp_pgno; - } + if ((rc = mdb_page_touch(mc))) + return rc; } return mdb_page_search_root(mc, key, modify); @@ -2833,7 +3593,7 @@ mdb_cursor_next(MDB_cursor *mc, MDB_val *key, MDB_val *data, MDB_cursor_op op) return rc; } } else { - mc->mc_xcursor->mx_cursor.mc_flags = 0; + mc->mc_xcursor->mx_cursor.mc_flags &= ~C_INITIALIZED; if (op == MDB_NEXT_DUP) return MDB_NOTFOUND; } @@ -2845,6 +3605,7 @@ mdb_cursor_next(MDB_cursor *mc, MDB_val *key, MDB_val *data, MDB_cursor_op op) DPUTS("=====> move to next sibling page"); if (mdb_cursor_sibling(mc, 1) != MDB_SUCCESS) { mc->mc_flags |= C_EOF; + mc->mc_flags &= ~C_INITIALIZED; return MDB_NOTFOUND; } mp = mc->mc_pg[mc->mc_top]; @@ -2902,7 +3663,7 @@ mdb_cursor_prev(MDB_cursor *mc, MDB_val *key, MDB_val *data, MDB_cursor_op op) if (op != MDB_PREV || rc == MDB_SUCCESS) return rc; } else { - mc->mc_xcursor->mx_cursor.mc_flags = 0; + mc->mc_xcursor->mx_cursor.mc_flags &= ~C_INITIALIZED; if (op == MDB_PREV_DUP) return MDB_NOTFOUND; } @@ -2974,6 +3735,10 @@ mdb_cursor_set(MDB_cursor *mc, MDB_val *key, MDB_val *data, MDB_val nodekey; mp = mc->mc_pg[mc->mc_top]; + if (!NUMKEYS(mp)) { + mc->mc_ki[mc->mc_top] = 0; + return MDB_NOTFOUND; + } if (mp->mp_flags & P_LEAF2) { nodekey.mv_size = mc->mc_db->md_pad; nodekey.mv_data = LEAF2KEY(mp, 0, nodekey.mv_size); @@ -3031,6 +3796,11 @@ mdb_cursor_set(MDB_cursor *mc, MDB_val *key, MDB_val *data, return MDB_NOTFOUND; } } + if (!mc->mc_top) { + /* There are no other pages */ + mc->mc_ki[mc->mc_top] = 0; + return MDB_NOTFOUND; + } } rc = mdb_page_search(mc, key, 0); @@ -3097,7 +3867,7 @@ set1: } else { if (mc->mc_xcursor) - mc->mc_xcursor->mx_cursor.mc_flags = 0; + mc->mc_xcursor->mx_cursor.mc_flags &= ~C_INITIALIZED; if ((rc = mdb_node_read(mc->mc_txn, leaf, data)) != MDB_SUCCESS) return rc; } @@ -3118,9 +3888,11 @@ mdb_cursor_first(MDB_cursor *mc, MDB_val *key, MDB_val *data) int rc; MDB_node *leaf; - rc = mdb_page_search(mc, NULL, 0); - if (rc != MDB_SUCCESS) - return rc; + if (!(mc->mc_flags & C_INITIALIZED) || mc->mc_top) { + rc = mdb_page_search(mc, NULL, 0); + if (rc != MDB_SUCCESS) + return rc; + } assert(IS_LEAF(mc->mc_pg[mc->mc_top])); leaf = NODEPTR(mc->mc_pg[mc->mc_top], 0); @@ -3143,7 +3915,7 @@ mdb_cursor_first(MDB_cursor *mc, MDB_val *key, MDB_val *data) return rc; } else { if (mc->mc_xcursor) - mc->mc_xcursor->mx_cursor.mc_flags = 0; + mc->mc_xcursor->mx_cursor.mc_flags &= ~C_INITIALIZED; if ((rc = mdb_node_read(mc->mc_txn, leaf, data)) != MDB_SUCCESS) return rc; } @@ -3163,9 +3935,11 @@ mdb_cursor_last(MDB_cursor *mc, MDB_val *key, MDB_val *data) lkey.mv_size = MAXKEYSIZE+1; lkey.mv_data = NULL; - rc = mdb_page_search(mc, &lkey, 0); - if (rc != MDB_SUCCESS) - return rc; + if (!(mc->mc_flags & C_INITIALIZED) || mc->mc_top) { + rc = mdb_page_search(mc, &lkey, 0); + if (rc != MDB_SUCCESS) + return rc; + } assert(IS_LEAF(mc->mc_pg[mc->mc_top])); leaf = NODEPTR(mc->mc_pg[mc->mc_top], NUMKEYS(mc->mc_pg[mc->mc_top])-1); @@ -3188,7 +3962,7 @@ mdb_cursor_last(MDB_cursor *mc, MDB_val *key, MDB_val *data) return rc; } else { if (mc->mc_xcursor) - mc->mc_xcursor->mx_cursor.mc_flags = 0; + mc->mc_xcursor->mx_cursor.mc_flags &= ~C_INITIALIZED; if ((rc = mdb_node_read(mc->mc_txn, leaf, data)) != MDB_SUCCESS) return rc; } @@ -3320,24 +4094,18 @@ mdb_cursor_touch(MDB_cursor *mc) { int rc; - if (mc->mc_dbi > MAIN_DBI && !mc->mc_dbx->md_dirty) { + if (mc->mc_dbi > MAIN_DBI && !(*mc->mc_dbflag & DB_DIRTY)) { MDB_cursor mc2; mdb_cursor_init(&mc2, mc->mc_txn, MAIN_DBI, NULL); rc = mdb_page_search(&mc2, &mc->mc_dbx->md_name, 1); if (rc) return rc; - mc->mc_dbx->md_dirty = 1; + *mc->mc_dbflag = DB_DIRTY; } for (mc->mc_top = 0; mc->mc_top < mc->mc_snum; mc->mc_top++) { - if (!F_ISSET(mc->mc_pg[mc->mc_top]->mp_flags, P_DIRTY)) { - rc = mdb_page_touch(mc); - if (rc) - return rc; - if (!mc->mc_top) { - mc->mc_db->md_root = - mc->mc_pg[mc->mc_top]->mp_pgno; - } - } + rc = mdb_page_touch(mc); + if (rc) + return rc; } mc->mc_top = mc->mc_snum-1; return MDB_SUCCESS; @@ -3347,20 +4115,24 @@ int mdb_cursor_put(MDB_cursor *mc, MDB_val *key, MDB_val *data, unsigned int flags) { - MDB_node *leaf; + MDB_node *leaf = NULL; MDB_val xdata, *rdata, dkey; + MDB_page *fp; MDB_db dummy; - char dbuf[PAGESIZE]; int do_sub = 0; + unsigned int mcount = 0; size_t nsize; - DKBUF; int rc, rc2; + MDB_pagebuf pbuf; + char dbuf[MAXKEYSIZE+1]; + unsigned int nflags; + DKBUF; if (F_ISSET(mc->mc_txn->mt_flags, MDB_TXN_RDONLY)) return EACCES; DPRINTF("==> put db %u key [%s], size %zu, data size %zu", - mc->mc_dbi, DKEY(key), key->mv_size, data->mv_size); + mc->mc_dbi, DKEY(key), key ? key->mv_size:0, data->mv_size); dkey.mv_size = 0; @@ -3379,7 +4151,7 @@ mdb_cursor_put(MDB_cursor *mc, MDB_val *key, MDB_val *data, mdb_cursor_push(mc, np); mc->mc_db->md_root = np->mp_pgno; mc->mc_db->md_depth++; - mc->mc_dbx->md_dirty = 1; + *mc->mc_dbflag = DB_DIRTY; if ((mc->mc_db->md_flags & (MDB_DUPSORT|MDB_DUPFIXED)) == MDB_DUPFIXED) np->mp_flags |= P_LEAF2; @@ -3390,7 +4162,7 @@ mdb_cursor_put(MDB_cursor *mc, MDB_val *key, MDB_val *data, int exact = 0; MDB_val d2; rc = mdb_cursor_set(mc, key, &d2, MDB_SET, &exact); - if (flags == MDB_NOOVERWRITE && rc == 0) { + if ((flags & MDB_NOOVERWRITE) && rc == 0) { DPRINTF("duplicate key [%s]", DKEY(key)); *data = d2; return MDB_KEYEXIST; @@ -3424,42 +4196,131 @@ top: /* DB has dups? */ if (F_ISSET(mc->mc_db->md_flags, MDB_DUPSORT)) { /* Was a single item before, must convert now */ +more: if (!F_ISSET(leaf->mn_flags, F_DUPDATA)) { + /* Just overwrite the current item */ + if (flags == MDB_CURRENT) + goto current; + dkey.mv_size = NODEDSZ(leaf); - dkey.mv_data = dbuf; - memcpy(dbuf, NODEDATA(leaf), dkey.mv_size); - /* data matches, ignore it */ + dkey.mv_data = NODEDATA(leaf); +#if UINT_MAX < SIZE_MAX + if (mc->mc_dbx->md_dcmp == mdb_cmp_int && dkey.mv_size == sizeof(size_t)) +#ifdef MISALIGNED_OK + mc->mc_dbx->md_dcmp = mdb_cmp_long; +#else + mc->mc_dbx->md_dcmp = mdb_cmp_cint; +#endif +#endif + /* if data matches, ignore it */ if (!mc->mc_dbx->md_dcmp(data, &dkey)) return (flags == MDB_NODUPDATA) ? MDB_KEYEXIST : MDB_SUCCESS; - memset(&dummy, 0, sizeof(dummy)); + + /* create a fake page for the dup items */ + memcpy(dbuf, dkey.mv_data, dkey.mv_size); + dkey.mv_data = dbuf; + fp = (MDB_page *)&pbuf; + fp->mp_pgno = mc->mc_pg[mc->mc_top]->mp_pgno; + fp->mp_flags = P_LEAF|P_DIRTY|P_SUBP; + fp->mp_lower = PAGEHDRSZ; + fp->mp_upper = PAGEHDRSZ + dkey.mv_size + data->mv_size; if (mc->mc_db->md_flags & MDB_DUPFIXED) { - dummy.md_pad = data->mv_size; - dummy.md_flags = MDB_DUPFIXED; - if (mc->mc_db->md_flags & MDB_INTEGERDUP) - dummy.md_flags |= MDB_INTEGERKEY; + fp->mp_flags |= P_LEAF2; + fp->mp_pad = data->mv_size; + } else { + fp->mp_upper += 2 * sizeof(indx_t) + 2 * NODESIZE + + (dkey.mv_size & 1) + (data->mv_size & 1); } - dummy.md_flags |= MDB_SUBDATA; - dummy.md_root = P_INVALID; - if (dkey.mv_size == sizeof(MDB_db)) { - memcpy(NODEDATA(leaf), &dummy, sizeof(dummy)); + mdb_node_del(mc->mc_pg[mc->mc_top], mc->mc_ki[mc->mc_top], 0); + do_sub = 1; + rdata = &xdata; + xdata.mv_size = fp->mp_upper; + xdata.mv_data = fp; + flags |= F_DUPDATA; + goto new_sub; + } + if (!F_ISSET(leaf->mn_flags, F_SUBDATA)) { + /* See if we need to convert from fake page to subDB */ + MDB_page *mp; + unsigned int offset; + unsigned int i; + + fp = NODEDATA(leaf); + if (flags == MDB_CURRENT) { + fp->mp_flags |= P_DIRTY; + COPY_PGNO(fp->mp_pgno, mc->mc_pg[mc->mc_top]->mp_pgno); + mc->mc_xcursor->mx_cursor.mc_pg[0] = fp; + flags |= F_DUPDATA; goto put_sub; } + if (mc->mc_db->md_flags & MDB_DUPFIXED) { + offset = fp->mp_pad; + } else { + offset = NODESIZE + sizeof(indx_t) + data->mv_size; + } + offset += offset & 1; + if (NODESIZE + sizeof(indx_t) + NODEKSZ(leaf) + NODEDSZ(leaf) + + offset >= (mc->mc_txn->mt_env->me_psize - PAGEHDRSZ) / + MDB_MINKEYS) { + /* yes, convert it */ + dummy.md_flags = 0; + if (mc->mc_db->md_flags & MDB_DUPFIXED) { + dummy.md_pad = fp->mp_pad; + dummy.md_flags = MDB_DUPFIXED; + if (mc->mc_db->md_flags & MDB_INTEGERDUP) + dummy.md_flags |= MDB_INTEGERKEY; + } + dummy.md_depth = 1; + dummy.md_branch_pages = 0; + dummy.md_leaf_pages = 1; + dummy.md_overflow_pages = 0; + dummy.md_entries = NUMKEYS(fp); + rdata = &xdata; + xdata.mv_size = sizeof(MDB_db); + xdata.mv_data = &dummy; + mp = mdb_page_alloc(mc, 1); + if (!mp) + return ENOMEM; + offset = mc->mc_txn->mt_env->me_psize - NODEDSZ(leaf); + flags |= F_DUPDATA|F_SUBDATA; + dummy.md_root = mp->mp_pgno; + } else { + /* no, just grow it */ + rdata = &xdata; + xdata.mv_size = NODEDSZ(leaf) + offset; + xdata.mv_data = &pbuf; + mp = (MDB_page *)&pbuf; + mp->mp_pgno = mc->mc_pg[mc->mc_top]->mp_pgno; + flags |= F_DUPDATA; + } + mp->mp_flags = fp->mp_flags | P_DIRTY; + mp->mp_pad = fp->mp_pad; + mp->mp_lower = fp->mp_lower; + mp->mp_upper = fp->mp_upper + offset; + if (IS_LEAF2(fp)) { + memcpy(METADATA(mp), METADATA(fp), NUMKEYS(fp) * fp->mp_pad); + } else { + nsize = NODEDSZ(leaf) - fp->mp_upper; + memcpy((char *)mp + mp->mp_upper, (char *)fp + fp->mp_upper, nsize); + for (i=0; imp_ptrs[i] = fp->mp_ptrs[i] + offset; + } mdb_node_del(mc->mc_pg[mc->mc_top], mc->mc_ki[mc->mc_top], 0); do_sub = 1; - rdata = &xdata; - xdata.mv_size = sizeof(MDB_db); - xdata.mv_data = &dummy; - /* new sub-DB, must fully init xcursor */ - if (flags == MDB_CURRENT) - flags = 0; goto new_sub; } + /* data is on sub-DB, just store it */ + flags |= F_DUPDATA|F_SUBDATA; goto put_sub; } +current: /* same size, just replace it */ if (!F_ISSET(leaf->mn_flags, F_BIGDATA) && NODEDSZ(leaf) == data->mv_size) { - memcpy(NODEDATA(leaf), data->mv_data, data->mv_size); + if (F_ISSET(flags, MDB_RESERVE)) + data->mv_data = NODEDATA(leaf); + else + memcpy(NODEDATA(leaf), data->mv_data, data->mv_size); goto done; } mdb_node_del(mc->mc_pg[mc->mc_top], mc->mc_ki[mc->mc_top], 0); @@ -3470,51 +4331,97 @@ top: rdata = data; new_sub: + nflags = flags & NODE_ADD_FLAGS; nsize = IS_LEAF2(mc->mc_pg[mc->mc_top]) ? key->mv_size : mdb_leaf_size(mc->mc_txn->mt_env, key, rdata); if (SIZELEFT(mc->mc_pg[mc->mc_top]) < nsize) { - rc = mdb_page_split(mc, key, rdata, P_INVALID); + if (( flags & (F_DUPDATA|F_SUBDATA)) == F_DUPDATA ) + nflags &= ~MDB_APPEND; + 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, 0); + rc = mdb_node_add(mc, mc->mc_ki[mc->mc_top], key, rdata, 0, nflags); + if (rc == 0 && !do_sub) { + /* Adjust other cursors pointing to mp */ + MDB_cursor *m2, *m3; + MDB_dbi dbi = mc->mc_dbi; + unsigned i = mc->mc_top; + MDB_page *mp = mc->mc_pg[i]; + + if (mc->mc_flags & C_SUB) + dbi--; + + for (m2 = mc->mc_txn->mt_cursors[dbi]; m2; m2=m2->mc_next) { + if (mc->mc_flags & C_SUB) + m3 = &m2->mc_xcursor->mx_cursor; + else + m3 = m2; + if (m3 == mc || m3->mc_snum < mc->mc_snum) continue; + if (m3->mc_pg[i] == mp && m3->mc_ki[i] >= mc->mc_ki[i]) { + m3->mc_ki[i]++; + } + } + } } if (rc != MDB_SUCCESS) mc->mc_txn->mt_flags |= MDB_TXN_ERROR; else { - /* Remember if we just added a subdatabase */ - if (flags & F_SUBDATA) { - leaf = NODEPTR(mc->mc_pg[mc->mc_top], mc->mc_ki[mc->mc_top]); - leaf->mn_flags |= F_SUBDATA; - } - /* 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) { - MDB_db *db; - leaf = NODEPTR(mc->mc_pg[mc->mc_top], mc->mc_ki[mc->mc_top]); + int xflags; put_sub: - if (flags != MDB_CURRENT) - mdb_xcursor_init1(mc, leaf); xdata.mv_size = 0; xdata.mv_data = ""; - if (flags == MDB_NODUPDATA) - flags = MDB_NOOVERWRITE; + leaf = NODEPTR(mc->mc_pg[mc->mc_top], mc->mc_ki[mc->mc_top]); + if (flags & MDB_CURRENT) { + xflags = MDB_CURRENT; + } else { + mdb_xcursor_init1(mc, leaf); + xflags = (flags & MDB_NODUPDATA) ? MDB_NOOVERWRITE : 0; + } /* converted, write the original data first */ if (dkey.mv_size) { - rc = mdb_cursor_put(&mc->mc_xcursor->mx_cursor, &dkey, &xdata, flags); + rc = mdb_cursor_put(&mc->mc_xcursor->mx_cursor, &dkey, &xdata, xflags); if (rc) return rc; - leaf->mn_flags |= F_DUPDATA; + { + /* 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_pg[i] == mp && m2->mc_ki[i] == mc->mc_ki[i]) { + mdb_xcursor_init1(m2, leaf); + } + } + } + } + xflags |= (flags & MDB_APPEND); + rc = mdb_cursor_put(&mc->mc_xcursor->mx_cursor, data, &xdata, xflags); + if (flags & F_SUBDATA) { + void *db = NODEDATA(leaf); + memcpy(db, &mc->mc_xcursor->mx_db, sizeof(MDB_db)); + } + } + /* 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)) + mc->mc_db->md_entries++; + if (flags & MDB_MULTIPLE) { + mcount++; + if (mcount < data[1].mv_size) { + data[0].mv_data = (char *)data[0].mv_data + data[0].mv_size; + leaf = NODEPTR(mc->mc_pg[mc->mc_top], mc->mc_ki[mc->mc_top]); + goto more; } - rc = mdb_cursor_put(&mc->mc_xcursor->mx_cursor, data, &xdata, flags); - db = NODEDATA(leaf); - assert((db->md_flags & MDB_SUBDATA) == MDB_SUBDATA); - memcpy(db, &mc->mc_xcursor->mx_db, sizeof(MDB_db)); } - mc->mc_db->md_entries++; } done: return rc; @@ -3540,48 +4447,33 @@ mdb_cursor_del(MDB_cursor *mc, unsigned int flags) if (!IS_LEAF2(mc->mc_pg[mc->mc_top]) && F_ISSET(leaf->mn_flags, F_DUPDATA)) { if (flags != MDB_NODUPDATA) { + if (!F_ISSET(leaf->mn_flags, F_SUBDATA)) { + mc->mc_xcursor->mx_cursor.mc_pg[0] = NODEDATA(leaf); + } rc = mdb_cursor_del(&mc->mc_xcursor->mx_cursor, 0); /* If sub-DB still has entries, we're done */ - if (mc->mc_xcursor->mx_db.md_root != P_INVALID) { - MDB_db *db = NODEDATA(leaf); - assert((db->md_flags & MDB_SUBDATA) == MDB_SUBDATA); - memcpy(db, &mc->mc_xcursor->mx_db, sizeof(MDB_db)); + if (mc->mc_xcursor->mx_db.md_entries) { + if (leaf->mn_flags & F_SUBDATA) { + /* update subDB info */ + void *db = NODEDATA(leaf); + memcpy(db, &mc->mc_xcursor->mx_db, sizeof(MDB_db)); + } else { + /* shrink fake page */ + mdb_node_shrink(mc->mc_pg[mc->mc_top], mc->mc_ki[mc->mc_top]); + } mc->mc_db->md_entries--; return rc; } /* otherwise fall thru and delete the sub-DB */ } - /* add all the child DB's pages to the free list */ - rc = mdb_page_search(&mc->mc_xcursor->mx_cursor, NULL, 0); - if (rc == MDB_SUCCESS) { - MDB_node *ni; - MDB_cursor *mx; - unsigned int i; - - mx = &mc->mc_xcursor->mx_cursor; - mc->mc_db->md_entries -= - mx->mc_db->md_entries; - - mdb_cursor_pop(mx); - while (mx->mc_snum > 1) { - for (i=0; imc_pg[mx->mc_top]); i++) { - MDB_page *mp; - pgno_t pg; - ni = NODEPTR(mx->mc_pg[mx->mc_top], i); - pg = NODEPGNO(ni); - if ((rc = mdb_page_get(mc->mc_txn, pg, &mp))) - return rc; - /* free it */ - mdb_midl_append(mc->mc_txn->mt_free_pgs, pg); - } - rc = mdb_cursor_sibling(mx, 1); - if (rc) - break; + 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; } - /* free it */ - mdb_midl_append(mc->mc_txn->mt_free_pgs, - mx->mc_db->md_root); } } @@ -3637,7 +4529,7 @@ mdb_leaf_size(MDB_env *env, MDB_val *key, MDB_val *data) size_t sz; sz = LEAFSIZE(key, data); - if (data->mv_size >= env->me_psize / MDB_MINKEYS) { + if (sz >= env->me_psize / MDB_MINKEYS) { /* put on overflow page */ sz -= data->mv_size - sizeof(pgno_t); } @@ -3688,7 +4580,7 @@ mdb_branch_size(MDB_env *env, MDB_val *key) */ static int mdb_node_add(MDB_cursor *mc, indx_t indx, - MDB_val *key, MDB_val *data, pgno_t pgno, uint8_t flags) + MDB_val *key, MDB_val *data, pgno_t pgno, unsigned int flags) { unsigned int i; size_t node_size = NODESIZE; @@ -3700,8 +4592,9 @@ mdb_node_add(MDB_cursor *mc, indx_t indx, assert(mp->mp_upper >= mp->mp_lower); - DPRINTF("add to %s page %zu index %i, data size %zu key size %zu [%s]", + DPRINTF("add to %s %spage %zu index %i, data size %zu key size %zu [%s]", IS_LEAF(mp) ? "leaf" : "branch", + IS_SUBP(mp) ? "sub-" : "", mp->mp_pgno, indx, data ? data->mv_size : 0, key ? key->mv_size : 0, key ? DKEY(key) : NULL); @@ -3729,11 +4622,11 @@ mdb_node_add(MDB_cursor *mc, indx_t indx, if (F_ISSET(flags, F_BIGDATA)) { /* Data already on overflow page. */ node_size += sizeof(pgno_t); - } else if (data->mv_size >= mc->mc_txn->mt_env->me_psize / MDB_MINKEYS) { + } else if (node_size + data->mv_size >= mc->mc_txn->mt_env->me_psize / MDB_MINKEYS) { int ovpages = OVPAGES(data->mv_size, mc->mc_txn->mt_env->me_psize); /* Put data on overflow page. */ - DPRINTF("data size is %zu, put on overflow page", - data->mv_size); + DPRINTF("data size is %zu, node would be %zu, put data on overflow page", + data->mv_size, node_size+data->mv_size); node_size += sizeof(pgno_t); if ((ofp = mdb_page_new(mc, P_OVERFLOW, ovpages)) == NULL) return ENOMEM; @@ -3783,13 +4676,18 @@ mdb_node_add(MDB_cursor *mc, indx_t indx, if (F_ISSET(flags, F_BIGDATA)) memcpy(node->mn_data + key->mv_size, data->mv_data, sizeof(pgno_t)); + else if (F_ISSET(flags, MDB_RESERVE)) + data->mv_data = node->mn_data + key->mv_size; else memcpy(node->mn_data + key->mv_size, data->mv_data, data->mv_size); } else { memcpy(node->mn_data + key->mv_size, &ofp->mp_pgno, sizeof(pgno_t)); - memcpy(METADATA(ofp), data->mv_data, data->mv_size); + if (F_ISSET(flags, MDB_RESERVE)) + data->mv_data = METADATA(ofp); + else + memcpy(METADATA(ofp), data->mv_data, data->mv_size); } } @@ -3810,8 +4708,14 @@ mdb_node_del(MDB_page *mp, indx_t indx, int ksize) MDB_node *node; char *base; +#if MDB_DEBUG + { + pgno_t pgno; + COPY_PGNO(pgno, mp->mp_pgno); DPRINTF("delete node %u on %s page %zu", indx, - IS_LEAF(mp) ? "leaf" : "branch", mp->mp_pgno); + IS_LEAF(mp) ? "leaf" : "branch", pgno); + } +#endif assert(indx < NUMKEYS(mp)); if (IS_LEAF2(mp)) { @@ -3852,6 +4756,58 @@ mdb_node_del(MDB_page *mp, indx_t indx, int ksize) mp->mp_upper += sz; } +/** Compact the main page after deleting a node on a subpage. + * @param[in] mp The main page to operate on. + * @param[in] indx The index of the subpage on the main page. + */ +static void +mdb_node_shrink(MDB_page *mp, indx_t indx) +{ + MDB_node *node; + MDB_page *sp, *xp; + char *base; + int osize, nsize; + int delta; + indx_t i, numkeys, ptr; + + node = NODEPTR(mp, indx); + sp = (MDB_page *)NODEDATA(node); + osize = NODEDSZ(node); + + delta = sp->mp_upper - sp->mp_lower; + SETDSZ(node, osize - delta); + xp = (MDB_page *)((char *)sp + delta); + + /* shift subpage upward */ + if (IS_LEAF2(sp)) { + nsize = NUMKEYS(sp) * sp->mp_pad; + memmove(METADATA(xp), METADATA(sp), nsize); + } else { + int i; + nsize = osize - sp->mp_upper; + numkeys = NUMKEYS(sp); + for (i=numkeys-1; i>=0; i--) + xp->mp_ptrs[i] = sp->mp_ptrs[i] - delta; + } + xp->mp_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); + + /* shift lower nodes upward */ + ptr = mp->mp_ptrs[indx]; + numkeys = NUMKEYS(mp); + for (i = 0; i < numkeys; i++) { + 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; +} + /** Initial setup of a sorted-dups cursor. * Sorted duplicates are implemented as a sub-database for the given key. * The duplicate data items are actually keys of the sub-database. @@ -3871,11 +4827,13 @@ mdb_xcursor_init0(MDB_cursor *mc) mx->mx_cursor.mc_db = &mx->mx_db; mx->mx_cursor.mc_dbx = &mx->mx_dbx; mx->mx_cursor.mc_dbi = mc->mc_dbi+1; - mx->mx_dbx.md_parent = mc->mc_dbi; + mx->mx_cursor.mc_dbflag = &mx->mx_dbflag; + mx->mx_cursor.mc_snum = 0; + mx->mx_cursor.mc_top = 0; + mx->mx_cursor.mc_flags = C_SUB; mx->mx_dbx.md_cmp = mc->mc_dbx->md_dcmp; mx->mx_dbx.md_dcmp = NULL; mx->mx_dbx.md_rel = mc->mc_dbx->md_rel; - mx->mx_dbx.md_dirty = 0; } /** Final setup of a sorted-dups cursor. @@ -3887,31 +4845,62 @@ mdb_xcursor_init0(MDB_cursor *mc) static void mdb_xcursor_init1(MDB_cursor *mc, MDB_node *node) { - MDB_db *db = NODEDATA(node); MDB_xcursor *mx = mc->mc_xcursor; - assert((db->md_flags & MDB_SUBDATA) == MDB_SUBDATA); - mx->mx_db = *db; + + if (node->mn_flags & F_SUBDATA) { + memcpy(&mx->mx_db, NODEDATA(node), sizeof(MDB_db)); + mx->mx_cursor.mc_snum = 0; + 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_flags = 0; + mx->mx_db.md_depth = 1; + mx->mx_db.md_branch_pages = 0; + mx->mx_db.md_leaf_pages = 1; + mx->mx_db.md_overflow_pages = 0; + mx->mx_db.md_entries = NUMKEYS(fp); + COPY_PGNO(mx->mx_db.md_root, fp->mp_pgno); + mx->mx_cursor.mc_snum = 1; + mx->mx_cursor.mc_flags = C_INITIALIZED|C_SUB; + mx->mx_cursor.mc_top = 0; + mx->mx_cursor.mc_pg[0] = fp; + mx->mx_cursor.mc_ki[0] = 0; + if (mc->mc_db->md_flags & MDB_DUPFIXED) { + mx->mx_db.md_flags = MDB_DUPFIXED; + mx->mx_db.md_pad = fp->mp_pad; + if (mc->mc_db->md_flags & MDB_INTEGERDUP) + mx->mx_db.md_flags |= MDB_INTEGERKEY; + } + } DPRINTF("Sub-db %u for db %u root page %zu", mx->mx_cursor.mc_dbi, mc->mc_dbi, - db->md_root); - if (F_ISSET(mc->mc_pg[mc->mc_top]->mp_flags, P_DIRTY)) - mx->mx_dbx.md_dirty = 1; + mx->mx_db.md_root); + mx->mx_dbflag = (F_ISSET(mc->mc_pg[mc->mc_top]->mp_flags, P_DIRTY)) ? + DB_DIRTY : 0; mx->mx_dbx.md_name.mv_data = NODEKEY(node); mx->mx_dbx.md_name.mv_size = node->mn_ksize; - mx->mx_cursor.mc_snum = 0; - mx->mx_cursor.mc_flags = 0; +#if UINT_MAX < SIZE_MAX if (mx->mx_dbx.md_cmp == mdb_cmp_int && mx->mx_db.md_pad == sizeof(size_t)) +#ifdef MISALIGNED_OK mx->mx_dbx.md_cmp = mdb_cmp_long; +#else + mx->mx_dbx.md_cmp = mdb_cmp_cint; +#endif +#endif } /** Initialize a cursor for a given transaction and database. */ static void mdb_cursor_init(MDB_cursor *mc, MDB_txn *txn, MDB_dbi dbi, MDB_xcursor *mx) { + mc->mc_orig = NULL; mc->mc_dbi = dbi; mc->mc_txn = txn; mc->mc_db = &txn->mt_dbs[dbi]; mc->mc_dbx = &txn->mt_dbxs[dbi]; + mc->mc_dbflag = &txn->mt_dbflags[dbi]; mc->mc_snum = 0; + mc->mc_top = 0; mc->mc_flags = 0; if (txn->mt_dbs[dbi].md_flags & MDB_DUPSORT) { assert(mx != NULL); @@ -3929,7 +4918,11 @@ mdb_cursor_open(MDB_txn *txn, MDB_dbi dbi, MDB_cursor **ret) MDB_xcursor *mx = NULL; size_t size = sizeof(MDB_cursor); - if (txn == NULL || ret == NULL || !dbi || dbi >= txn->mt_numdbs) + if (txn == NULL || ret == NULL || dbi >= txn->mt_numdbs) + return EINVAL; + + /* Allow read access to the freelist */ + if (!dbi && !F_ISSET(txn->mt_flags, MDB_TXN_RDONLY)) return EINVAL; if (txn->mt_dbs[dbi].md_flags & MDB_DUPSORT) @@ -3940,6 +4933,11 @@ mdb_cursor_open(MDB_txn *txn, MDB_dbi dbi, MDB_cursor **ret) mx = (MDB_xcursor *)(mc + 1); } mdb_cursor_init(mc, txn, dbi, mx); + if (txn->mt_cursors) { + mc->mc_next = txn->mt_cursors[dbi]; + txn->mt_cursors[dbi] = mc; + } + mc->mc_flags |= C_ALLOCD; } else { return ENOMEM; } @@ -3977,10 +4975,32 @@ void mdb_cursor_close(MDB_cursor *mc) { if (mc != NULL) { - free(mc); + /* remove from txn, if tracked */ + if (mc->mc_txn->mt_cursors) { + MDB_cursor **prev = &mc->mc_txn->mt_cursors[mc->mc_dbi]; + while (*prev && *prev != mc) prev = &(*prev)->mc_next; + if (*prev == mc) + *prev = mc->mc_next; + } + if (mc->mc_flags & C_ALLOCD) + free(mc); } } +MDB_txn * +mdb_cursor_txn(MDB_cursor *mc) +{ + if (!mc) return NULL; + return mc->mc_txn; +} + +MDB_dbi +mdb_cursor_dbi(MDB_cursor *mc) +{ + if (!mc) return 0; + return mc->mc_dbi; +} + /** Replace the key for a node with a new key. * @param[in] mp The page containing the node to operate on. * @param[in] indx The index of the node to operate on. @@ -3999,11 +5019,19 @@ mdb_update_key(MDB_page *mp, indx_t indx, MDB_val *key) node = NODEPTR(mp, indx); ptr = mp->mp_ptrs[indx]; - DPRINTF("update key %u (ofs %u) [%.*s] to [%s] on page %zu", - indx, ptr, - (int)node->mn_ksize, (char *)NODEKEY(node), - DKEY(key), - mp->mp_pgno); +#if MDB_DEBUG + { + MDB_val k2; + char kbuf2[(MAXKEYSIZE*2+1)]; + k2.mv_data = NODEKEY(node); + k2.mv_size = node->mn_ksize; + DPRINTF("update key %u (ofs %u) [%s] to [%s] on page %zu", + indx, ptr, + mdb_dkey(&k2, kbuf2), + DKEY(key), + mp->mp_pgno); + } +#endif delta = key->mv_size - node->mn_ksize; if (delta) { @@ -4027,7 +5055,8 @@ mdb_update_key(MDB_page *mp, indx_t indx, MDB_val *key) node->mn_ksize = key->mv_size; } - memcpy(NODEKEY(node), key->mv_data, key->mv_size); + if (key->mv_size) + memcpy(NODEKEY(node), key->mv_data, key->mv_size); return MDB_SUCCESS; } @@ -4040,6 +5069,9 @@ mdb_node_move(MDB_cursor *csrc, MDB_cursor *cdst) int rc; MDB_node *srcnode; MDB_val key, data; + pgno_t srcpg; + unsigned short flags; + DKBUF; /* Mark src and dst as dirty. */ @@ -4053,8 +5085,13 @@ mdb_node_move(MDB_cursor *csrc, MDB_cursor *cdst) key.mv_data = LEAF2KEY(csrc->mc_pg[csrc->mc_top], csrc->mc_ki[csrc->mc_top], key.mv_size); data.mv_size = 0; data.mv_data = NULL; + srcpg = 0; + flags = 0; } else { srcnode = NODEPTR(csrc->mc_pg[csrc->mc_top], csrc->mc_ki[csrc->mc_top]); + assert(!((long)srcnode&1)); + srcpg = NODEPGNO(srcnode); + flags = srcnode->mn_flags; if (csrc->mc_ki[csrc->mc_top] == 0 && IS_BRANCH(csrc->mc_pg[csrc->mc_top])) { unsigned int snum = csrc->mc_snum; MDB_node *s2; @@ -4072,6 +5109,20 @@ mdb_node_move(MDB_cursor *csrc, MDB_cursor *cdst) data.mv_size = NODEDSZ(srcnode); data.mv_data = NODEDATA(srcnode); } + if (IS_BRANCH(cdst->mc_pg[cdst->mc_top]) && cdst->mc_ki[cdst->mc_top] == 0) { + unsigned int snum = cdst->mc_snum; + MDB_node *s2; + MDB_val bkey; + /* must find the lowest key below dst */ + mdb_page_search_root(cdst, NULL, 0); + s2 = NODEPTR(cdst->mc_pg[cdst->mc_top], 0); + bkey.mv_size = NODEKSZ(s2); + bkey.mv_data = NODEKEY(s2); + cdst->mc_snum = snum--; + cdst->mc_top = snum; + rc = mdb_update_key(cdst->mc_pg[cdst->mc_top], 0, &bkey); + } + DPRINTF("moving %s node %u [%s] on page %zu to node %u on page %zu", IS_LEAF(csrc->mc_pg[csrc->mc_top]) ? "leaf" : "branch", csrc->mc_ki[csrc->mc_top], @@ -4081,8 +5132,7 @@ mdb_node_move(MDB_cursor *csrc, MDB_cursor *cdst) /* Add the node to the destination page. */ - rc = mdb_node_add(cdst, cdst->mc_ki[cdst->mc_top], &key, &data, NODEPGNO(srcnode), - srcnode->mn_flags); + rc = mdb_node_add(cdst, cdst->mc_ki[cdst->mc_top], &key, &data, srcpg, flags); if (rc != MDB_SUCCESS) return rc; @@ -4090,6 +5140,29 @@ mdb_node_move(MDB_cursor *csrc, MDB_cursor *cdst) */ mdb_node_del(csrc->mc_pg[csrc->mc_top], csrc->mc_ki[csrc->mc_top], key.mv_size); + { + /* 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]; + + if (csrc->mc_flags & C_SUB) + dbi--; + + for (m2 = csrc->mc_txn->mt_cursors[dbi]; m2; m2=m2->mc_next) { + if (m2 == csrc) continue; + if (csrc->mc_flags & C_SUB) + m3 = &m2->mc_xcursor->mx_cursor; + else + m3 = m2; + if (m3->mc_pg[csrc->mc_top] == mp && m3->mc_ki[csrc->mc_top] == + csrc->mc_ki[csrc->mc_top]) { + m3->mc_pg[csrc->mc_top] = cdst->mc_pg[cdst->mc_top]; + m3->mc_ki[csrc->mc_top] = cdst->mc_ki[cdst->mc_top]; + } + } + } + /* Update the parent separators. */ if (csrc->mc_ki[csrc->mc_top] == 0) { @@ -4110,7 +5183,8 @@ mdb_node_move(MDB_cursor *csrc, MDB_cursor *cdst) if (IS_BRANCH(csrc->mc_pg[csrc->mc_top])) { MDB_val nullkey; nullkey.mv_size = 0; - assert(mdb_update_key(csrc->mc_pg[csrc->mc_top], 0, &nullkey) == MDB_SUCCESS); + rc = mdb_update_key(csrc->mc_pg[csrc->mc_top], 0, &nullkey); + assert(rc == MDB_SUCCESS); } } @@ -4132,7 +5206,8 @@ mdb_node_move(MDB_cursor *csrc, MDB_cursor *cdst) if (IS_BRANCH(cdst->mc_pg[cdst->mc_top])) { MDB_val nullkey; nullkey.mv_size = 0; - assert(mdb_update_key(cdst->mc_pg[cdst->mc_top], 0, &nullkey) == MDB_SUCCESS); + rc = mdb_update_key(cdst->mc_pg[cdst->mc_top], 0, &nullkey); + assert(rc == MDB_SUCCESS); } } @@ -4153,6 +5228,7 @@ mdb_page_merge(MDB_cursor *csrc, MDB_cursor *cdst) indx_t i, j; MDB_node *srcnode; MDB_val key, data; + unsigned nkeys; DPRINTF("merging page %zu into %zu", csrc->mc_pg[csrc->mc_top]->mp_pgno, cdst->mc_pg[cdst->mc_top]->mp_pgno); @@ -4166,7 +5242,7 @@ mdb_page_merge(MDB_cursor *csrc, MDB_cursor *cdst) /* Move all nodes from src to dst. */ - j = NUMKEYS(cdst->mc_pg[cdst->mc_top]); + j = nkeys = NUMKEYS(cdst->mc_pg[cdst->mc_top]); if (IS_LEAF2(csrc->mc_pg[csrc->mc_top])) { key.mv_size = csrc->mc_db->md_pad; key.mv_data = METADATA(csrc->mc_pg[csrc->mc_top]); @@ -4179,9 +5255,21 @@ mdb_page_merge(MDB_cursor *csrc, MDB_cursor *cdst) } 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; + MDB_node *s2; + /* must find the lowest key below src */ + mdb_page_search_root(csrc, NULL, 0); + s2 = NODEPTR(csrc->mc_pg[csrc->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); + } - key.mv_size = srcnode->mn_ksize; - key.mv_data = NODEKEY(srcnode); data.mv_size = NODEDSZ(srcnode); data.mv_data = NODEDATA(srcnode); rc = mdb_node_add(cdst, j, &key, &data, NODEPGNO(srcnode), srcnode->mn_flags); @@ -4202,11 +5290,33 @@ mdb_page_merge(MDB_cursor *csrc, MDB_cursor *cdst) return rc; } - mdb_midl_append(csrc->mc_txn->mt_free_pgs, csrc->mc_pg[csrc->mc_top]->mp_pgno); + mdb_midl_append(&csrc->mc_txn->mt_free_pgs, csrc->mc_pg[csrc->mc_top]->mp_pgno); if (IS_LEAF(csrc->mc_pg[csrc->mc_top])) 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]; + + if (csrc->mc_flags & C_SUB) + dbi--; + + 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 == 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; + m3->mc_ki[csrc->mc_top] += nkeys; + } + } + } mdb_cursor_pop(csrc); return mdb_rebalance(csrc); @@ -4248,33 +5358,86 @@ mdb_rebalance(MDB_cursor *mc) unsigned int ptop; MDB_cursor mn; +#if MDB_DEBUG + { + pgno_t pgno; + COPY_PGNO(pgno, mc->mc_pg[mc->mc_top]->mp_pgno); DPRINTF("rebalancing %s page %zu (has %u keys, %.1f%% full)", IS_LEAF(mc->mc_pg[mc->mc_top]) ? "leaf" : "branch", - mc->mc_pg[mc->mc_top]->mp_pgno, NUMKEYS(mc->mc_pg[mc->mc_top]), (float)PAGEFILL(mc->mc_txn->mt_env, mc->mc_pg[mc->mc_top]) / 10); + pgno, NUMKEYS(mc->mc_pg[mc->mc_top]), (float)PAGEFILL(mc->mc_txn->mt_env, mc->mc_pg[mc->mc_top]) / 10); + } +#endif if (PAGEFILL(mc->mc_txn->mt_env, mc->mc_pg[mc->mc_top]) >= FILL_THRESHOLD) { +#if MDB_DEBUG + pgno_t pgno; + COPY_PGNO(pgno, mc->mc_pg[mc->mc_top]->mp_pgno); DPRINTF("no need to rebalance page %zu, above fill threshold", - mc->mc_pg[mc->mc_top]->mp_pgno); + pgno); +#endif return MDB_SUCCESS; } if (mc->mc_snum < 2) { - if (NUMKEYS(mc->mc_pg[mc->mc_top]) == 0) { + MDB_page *mp = mc->mc_pg[0]; + if (NUMKEYS(mp) == 0) { DPUTS("tree is completely empty"); mc->mc_db->md_root = P_INVALID; mc->mc_db->md_depth = 0; mc->mc_db->md_leaf_pages = 0; - mdb_midl_append(mc->mc_txn->mt_free_pgs, mc->mc_pg[mc->mc_top]->mp_pgno); + mdb_midl_append(&mc->mc_txn->mt_free_pgs, mp->mp_pgno); mc->mc_snum = 0; - } else if (IS_BRANCH(mc->mc_pg[mc->mc_top]) && NUMKEYS(mc->mc_pg[mc->mc_top]) == 1) { + mc->mc_top = 0; + { + /* Adjust other cursors pointing to mp */ + MDB_cursor *m2, *m3; + MDB_dbi dbi = mc->mc_dbi; + + if (mc->mc_flags & C_SUB) + dbi--; + + for (m2 = mc->mc_txn->mt_cursors[dbi]; m2; m2=m2->mc_next) { + if (m2 == mc) continue; + if (mc->mc_flags & C_SUB) + m3 = &m2->mc_xcursor->mx_cursor; + else + m3 = m2; + if (m3->mc_snum < mc->mc_snum) continue; + if (m3->mc_pg[0] == mp) { + m3->mc_snum = 0; + m3->mc_top = 0; + } + } + } + } else if (IS_BRANCH(mp) && NUMKEYS(mp) == 1) { DPUTS("collapsing root page!"); - mdb_midl_append(mc->mc_txn->mt_free_pgs, mc->mc_pg[mc->mc_top]->mp_pgno); - mc->mc_db->md_root = NODEPGNO(NODEPTR(mc->mc_pg[mc->mc_top], 0)); + mdb_midl_append(&mc->mc_txn->mt_free_pgs, mp->mp_pgno); + mc->mc_db->md_root = NODEPGNO(NODEPTR(mp, 0)); if ((rc = mdb_page_get(mc->mc_txn, mc->mc_db->md_root, - &mc->mc_pg[mc->mc_top]))) + &mc->mc_pg[0]))) return rc; mc->mc_db->md_depth--; mc->mc_db->md_branch_pages--; + { + /* Adjust other cursors pointing to mp */ + MDB_cursor *m2, *m3; + MDB_dbi dbi = mc->mc_dbi; + + if (mc->mc_flags & C_SUB) + dbi--; + + for (m2 = mc->mc_txn->mt_cursors[dbi]; m2; m2=m2->mc_next) { + if (m2 == mc) continue; + if (mc->mc_flags & C_SUB) + m3 = &m2->mc_xcursor->mx_cursor; + else + m3 = m2; + if (m3->mc_snum < mc->mc_snum) continue; + if (m3->mc_pg[0] == mp) { + m3->mc_pg[0] = mc->mc_pg[0]; + } + } + } } else DPUTS("root page doesn't need rebalancing"); return MDB_SUCCESS; @@ -4329,6 +5492,7 @@ mdb_rebalance(MDB_cursor *mc) if (PAGEFILL(mc->mc_txn->mt_env, mn.mc_pg[mn.mc_top]) >= FILL_THRESHOLD && NUMKEYS(mn.mc_pg[mn.mc_top]) >= 2) return mdb_node_move(&mn, mc); else { /* FIXME: if (has_enough_room()) */ + mc->mc_flags &= ~C_INITIALIZED; if (mc->mc_ki[ptop] == 0) return mdb_page_merge(&mn, mc); else @@ -4349,9 +5513,10 @@ mdb_cursor_del0(MDB_cursor *mc, MDB_node *leaf) memcpy(&pg, NODEDATA(leaf), sizeof(pg)); ovpages = OVPAGES(NODEDSZ(leaf), mc->mc_txn->mt_env->me_psize); + mc->mc_db->md_overflow_pages -= ovpages; for (i=0; imc_txn->mt_free_pgs, pg); + mdb_midl_append(&mc->mc_txn->mt_free_pgs, pg); pg++; } } @@ -4417,15 +5582,16 @@ mdb_del(MDB_txn *txn, MDB_dbi dbi, * @return 0 on success, non-zero on failure. */ static int -mdb_page_split(MDB_cursor *mc, MDB_val *newkey, MDB_val *newdata, pgno_t newpgno) +mdb_page_split(MDB_cursor *mc, MDB_val *newkey, MDB_val *newdata, pgno_t newpgno, + unsigned int nflags) { - uint8_t flags; - int rc = MDB_SUCCESS, ins_new = 0; + unsigned int flags; + int rc = MDB_SUCCESS, ins_new = 0, new_root = 0; indx_t newindx; pgno_t pgno = 0; unsigned int i, j, split_indx, nkeys, pmax; MDB_node *node; - MDB_val sepkey, rkey, rdata; + MDB_val sepkey, rkey, xdata, *rdata = &xdata; MDB_page *copy; MDB_page *mp, *rp, *pp; unsigned int ptop; @@ -4439,6 +5605,11 @@ mdb_page_split(MDB_cursor *mc, MDB_val *newkey, MDB_val *newdata, pgno_t newpgno IS_LEAF(mp) ? "leaf" : "branch", mp->mp_pgno, DKEY(newkey), mc->mc_ki[mc->mc_top]); + /* Create a right sibling. */ + if ((rp = mdb_page_new(mc, mp->mp_flags, 1)) == NULL) + return ENOMEM; + DPRINTF("new right sibling: page %zu", rp->mp_pgno); + if (mc->mc_snum < 2) { if ((pp = mdb_page_new(mc, P_BRANCH, 1)) == NULL) return ENOMEM; @@ -4450,6 +5621,7 @@ mdb_page_split(MDB_cursor *mc, MDB_val *newkey, MDB_val *newdata, pgno_t newpgno mc->mc_db->md_root = pp->mp_pgno; DPRINTF("root split! new root = %zu", pp->mp_pgno); mc->mc_db->md_depth++; + new_root = 1; /* Add left (implicit) pointer. */ if ((rc = mdb_node_add(mc, 0, NULL, NULL, mp->mp_pgno, 0)) != MDB_SUCCESS) { @@ -4468,13 +5640,17 @@ mdb_page_split(MDB_cursor *mc, MDB_val *newkey, MDB_val *newdata, pgno_t newpgno DPRINTF("parent branch page is %zu", mc->mc_pg[ptop]->mp_pgno); } - /* Create a right sibling. */ - if ((rp = mdb_page_new(mc, mp->mp_flags, 1)) == NULL) - return ENOMEM; mdb_cursor_copy(mc, &mn); mn.mc_pg[mn.mc_top] = rp; mn.mc_ki[ptop] = mc->mc_ki[ptop]+1; - DPRINTF("new right sibling: page %zu", rp->mp_pgno); + + if (nflags & MDB_APPEND) { + mn.mc_ki[mn.mc_top] = 0; + sepkey = *newkey; + nkeys = 0; + split_indx = 0; + goto newsep; + } nkeys = NUMKEYS(mp); split_indx = nkeys / 2 + 1; @@ -4582,7 +5758,7 @@ newsep: if (SIZELEFT(mn.mc_pg[ptop]) < mdb_branch_size(mc->mc_txn->mt_env, &sepkey)) { mn.mc_snum--; mn.mc_top--; - rc = mdb_page_split(&mn, &sepkey, NULL, rp->mp_pgno); + rc = mdb_page_split(&mn, &sepkey, NULL, rp->mp_pgno, 0); /* Right page might now have changed parent. * Check if left page also changed parent. @@ -4597,23 +5773,27 @@ newsep: rc = mdb_node_add(&mn, mn.mc_ki[ptop], &sepkey, NULL, rp->mp_pgno, 0); mn.mc_top++; } - if (IS_LEAF2(rp)) { - return rc; - } if (rc != MDB_SUCCESS) { return rc; } + 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; + } + if (IS_LEAF2(rp)) { + goto done; + } /* Move half of the keys to the right sibling. */ /* grab a page to hold a temporary copy */ - if (mc->mc_txn->mt_env->me_dpages) { - copy = mc->mc_txn->mt_env->me_dpages; - mc->mc_txn->mt_env->me_dpages = copy->mp_next; - } else { - if ((copy = malloc(mc->mc_txn->mt_env->me_psize)) == NULL) - return ENOMEM; - } + copy = mdb_page_malloc(mc); + if (copy == NULL) + return ENOMEM; copy->mp_pgno = mp->mp_pgno; copy->mp_flags = mp->mp_flags; @@ -4633,15 +5813,16 @@ newsep: rkey.mv_data = newkey->mv_data; rkey.mv_size = newkey->mv_size; if (IS_LEAF(mp)) { - rdata.mv_data = newdata->mv_data; - rdata.mv_size = newdata->mv_size; + rdata = newdata; } else pgno = newpgno; - flags = 0; + flags = nflags; ins_new = 1; /* Update page and index for the new key. */ + if (!newindx) + mc->mc_pg[mc->mc_top] = copy; mc->mc_ki[mc->mc_top] = j; } else if (i == nkeys) { break; @@ -4650,8 +5831,9 @@ newsep: rkey.mv_data = NODEKEY(node); rkey.mv_size = node->mn_ksize; if (IS_LEAF(mp)) { - rdata.mv_data = NODEDATA(node); - rdata.mv_size = NODEDSZ(node); + xdata.mv_data = NODEDATA(node); + xdata.mv_size = NODEDSZ(node); + rdata = &xdata; } else pgno = NODEPGNO(node); flags = node->mn_flags; @@ -4664,13 +5846,9 @@ newsep: rkey.mv_size = 0; } - rc = mdb_node_add(mc, j, &rkey, &rdata, pgno, flags); + rc = mdb_node_add(mc, j, &rkey, rdata, pgno, flags); } - /* reset back to original page */ - if (newindx < split_indx) - mc->mc_pg[mc->mc_top] = mp; - nkeys = NUMKEYS(copy); for (i=0; imp_ptrs[i] = copy->mp_ptrs[i]; @@ -4679,9 +5857,57 @@ newsep: memcpy(NODEPTR(mp, nkeys-1), NODEPTR(copy, nkeys-1), mc->mc_txn->mt_env->me_psize - copy->mp_upper); + /* reset back to original page */ + if (!newindx || (newindx < split_indx)) { + mc->mc_pg[mc->mc_top] = mp; + if (nflags & MDB_RESERVE) { + node = NODEPTR(mp, mc->mc_ki[mc->mc_top]); + if (!(node->mn_flags & F_BIGDATA)) + newdata->mv_data = NODEDATA(node); + } + } + /* return tmp page to freelist */ copy->mp_next = mc->mc_txn->mt_env->me_dpages; + VGMEMP_FREE(mc->mc_txn->mt_env, copy); mc->mc_txn->mt_env->me_dpages = copy; +done: + { + /* Adjust other cursors pointing to mp */ + MDB_cursor *m2, *m3; + MDB_dbi dbi = mc->mc_dbi; + + if (mc->mc_flags & C_SUB) + dbi--; + + for (m2 = mc->mc_txn->mt_cursors[dbi]; m2; m2=m2->mc_next) { + if (m2 == mc) continue; + if (mc->mc_flags & C_SUB) + m3 = &m2->mc_xcursor->mx_cursor; + else + m3 = m2; + if (!(m3->mc_flags & C_INITIALIZED)) + continue; + if (new_root) { + int k; + /* root split */ + for (k=m3->mc_top; k>=0; k--) { + m3->mc_ki[k+1] = m3->mc_ki[k]; + m3->mc_pg[k+1] = m3->mc_pg[k]; + } + m3->mc_ki[0] = mc->mc_ki[0]; + m3->mc_pg[0] = mc->mc_pg[0]; + m3->mc_snum++; + m3->mc_top++; + } + if (m3->mc_pg[mc->mc_top] == mp) { + if (m3->mc_ki[m3->mc_top] >= split_indx) { + m3->mc_pg[m3->mc_top] = rp; + m3->mc_ki[m3->mc_top] -= split_indx; + } + } + } + } return rc; } @@ -4706,21 +5932,21 @@ mdb_put(MDB_txn *txn, MDB_dbi dbi, return EINVAL; } - if ((flags & (MDB_NOOVERWRITE|MDB_NODUPDATA)) != flags) + if ((flags & (MDB_NOOVERWRITE|MDB_NODUPDATA|MDB_RESERVE|MDB_APPEND)) != flags) return EINVAL; mdb_cursor_init(&mc, txn, dbi, &mx); return mdb_cursor_put(&mc, key, data, flags); } +/** Only a subset of the @ref mdb_env flags can be changed + * at runtime. Changing other flags requires closing the environment + * and re-opening it with the new flags. + */ +#define CHANGEABLE (MDB_NOSYNC) int mdb_env_set_flags(MDB_env *env, unsigned int flag, int onoff) { - /** Only a subset of the @ref mdb_env flags can be changed - * at runtime. Changing other flags requires closing the environment - * and re-opening it with the new flags. - */ -#define CHANGEABLE (MDB_NOSYNC) if ((flag & CHANGEABLE) != flag) return EINVAL; if (onoff) @@ -4818,7 +6044,8 @@ int mdb_open(MDB_txn *txn, const char *name, unsigned int flags, MDB_dbi *dbi) { MDB_val key, data; MDB_dbi i; - int rc, dirty = 0; + MDB_cursor mc; + int rc, dbflag, exact; size_t len; if (txn->mt_dbxs[FREE_DBI].md_cmp == NULL) { @@ -4852,22 +6079,27 @@ int mdb_open(MDB_txn *txn, const char *name, unsigned int flags, MDB_dbi *dbi) return ENFILE; /* Find the DB info */ + dbflag = 0; + exact = 0; key.mv_size = len; key.mv_data = (void *)name; - rc = mdb_get(txn, MAIN_DBI, &key, &data); - - /* Create if requested */ - if (rc == MDB_NOTFOUND && (flags & MDB_CREATE)) { - MDB_cursor mc; + mdb_cursor_init(&mc, txn, MAIN_DBI, NULL); + rc = mdb_cursor_set(&mc, &key, &data, MDB_SET, &exact); + 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)) + return EINVAL; + } 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)); dummy.md_root = P_INVALID; dummy.md_flags = flags & 0xffff; - mdb_cursor_init(&mc, txn, MAIN_DBI, NULL); rc = mdb_cursor_put(&mc, &key, &data, F_SUBDATA); - dirty = 1; + dbflag = DB_DIRTY; } /* OK, got info, add to table */ @@ -4875,8 +6107,7 @@ int mdb_open(MDB_txn *txn, const char *name, unsigned int flags, MDB_dbi *dbi) txn->mt_dbxs[txn->mt_numdbs].md_name.mv_data = strdup(name); txn->mt_dbxs[txn->mt_numdbs].md_name.mv_size = len; txn->mt_dbxs[txn->mt_numdbs].md_rel = NULL; - txn->mt_dbxs[txn->mt_numdbs].md_parent = MAIN_DBI; - txn->mt_dbxs[txn->mt_numdbs].md_dirty = dirty; + txn->mt_dbflags[txn->mt_numdbs] = dbflag; memcpy(&txn->mt_dbs[txn->mt_numdbs], data.mv_data, sizeof(MDB_db)); *dbi = txn->mt_numdbs; txn->mt_env->me_dbs[0][txn->mt_numdbs] = txn->mt_dbs[txn->mt_numdbs]; @@ -4896,17 +6127,112 @@ int mdb_stat(MDB_txn *txn, MDB_dbi dbi, MDB_stat *arg) return mdb_stat0(txn->mt_env, &txn->mt_dbs[dbi], arg); } -void mdb_close(MDB_txn *txn, MDB_dbi dbi) +void mdb_close(MDB_env *env, MDB_dbi dbi) { char *ptr; - if (dbi <= MAIN_DBI || dbi >= txn->mt_numdbs) + if (dbi <= MAIN_DBI || dbi >= env->me_numdbs) return; - ptr = txn->mt_dbxs[dbi].md_name.mv_data; - txn->mt_dbxs[dbi].md_name.mv_data = NULL; - txn->mt_dbxs[dbi].md_name.mv_size = 0; + 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; free(ptr); } +/** Add all the DB's pages to the free list. + * @param[in] mc Cursor on the DB to free. + * @param[in] subs non-Zero to check for sub-DBs in this DB. + * @return 0 on success, non-zero on failure. + */ +static int +mdb_drop0(MDB_cursor *mc, int subs) +{ + int rc; + + rc = mdb_page_search(mc, NULL, 0); + if (rc == MDB_SUCCESS) { + MDB_node *ni; + MDB_cursor mx; + unsigned int i; + + /* LEAF2 pages have no nodes, cannot have sub-DBs */ + if (!subs || IS_LEAF2(mc->mc_pg[mc->mc_top])) + mdb_cursor_pop(mc); + + mdb_cursor_copy(mc, &mx); + while (mc->mc_snum > 0) { + if (IS_LEAF(mc->mc_pg[mc->mc_top])) { + for (i=0; imc_pg[mc->mc_top]); i++) { + ni = NODEPTR(mc->mc_pg[mc->mc_top], i); + if (ni->mn_flags & F_SUBDATA) { + mdb_xcursor_init1(mc, ni); + rc = mdb_drop0(&mc->mc_xcursor->mx_cursor, 0); + if (rc) + return rc; + } + } + } else { + for (i=0; imc_pg[mc->mc_top]); i++) { + pgno_t pg; + ni = NODEPTR(mc->mc_pg[mc->mc_top], i); + pg = NODEPGNO(ni); + /* free it */ + mdb_midl_append(&mc->mc_txn->mt_free_pgs, pg); + } + } + if (!mc->mc_top) + break; + rc = mdb_cursor_sibling(mc, 1); + if (rc) { + /* no more siblings, go back to beginning + * of previous level. (stack was already popped + * by mdb_cursor_sibling) + */ + for (i=1; imc_top; i++) + mc->mc_pg[i] = mx.mc_pg[i]; + } + } + /* free it */ + mdb_midl_append(&mc->mc_txn->mt_free_pgs, + mc->mc_db->md_root); + } + return 0; +} + +int mdb_drop(MDB_txn *txn, MDB_dbi dbi, int del) +{ + MDB_cursor *mc; + int rc; + + if (!txn || !dbi || dbi >= txn->mt_numdbs) + return EINVAL; + + rc = mdb_cursor_open(txn, dbi, &mc); + if (rc) + return rc; + + rc = mdb_drop0(mc, mc->mc_db->md_flags & MDB_DUPSORT); + if (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 (!rc) + mdb_close(txn->mt_env, dbi); + } else { + txn->mt_dbflags[dbi] |= DB_DIRTY; + txn->mt_dbs[dbi].md_depth = 0; + txn->mt_dbs[dbi].md_branch_pages = 0; + txn->mt_dbs[dbi].md_leaf_pages = 0; + txn->mt_dbs[dbi].md_overflow_pages = 0; + txn->mt_dbs[dbi].md_entries = 0; + txn->mt_dbs[dbi].md_root = P_INVALID; + } +leave: + mdb_cursor_close(mc); + return rc; +} + int mdb_set_compare(MDB_txn *txn, MDB_dbi dbi, MDB_cmp_func *cmp) { if (txn == NULL || !dbi || dbi >= txn->mt_numdbs)