X-Git-Url: https://git.sur5r.net/?a=blobdiff_plain;f=libraries%2Flibmdb%2Fmdb.c;h=2796d2740b893de8e7a30beca58a9ab1287df816;hb=f9c8796d0b3ed9bc0f51c76bb28609121b1e2eec;hp=14a4c21b275e69664b435b9e847913bb92fc1be7;hpb=94718fab000c63c0bd06e43bf91b9b0a3788ccd7;p=openldap diff --git a/libraries/libmdb/mdb.c b/libraries/libmdb/mdb.c index 14a4c21b27..2796d2740b 100644 --- a/libraries/libmdb/mdb.c +++ b/libraries/libmdb/mdb.c @@ -1,4 +1,9 @@ -/* mdb.c - memory-mapped database library */ +/** @file mdb.c + * @brief memory-mapped database library + * + * A Btree-based database management library modeled loosely on the + * BerkeleyDB API, but much simplified. + */ /* * Copyright 2011 Howard Chu, Symas Corp. * All rights reserved. @@ -30,12 +35,16 @@ #include #include #include +#ifdef _WIN32 +#include +#else #include #include #ifdef HAVE_SYS_FILE_H #include #endif #include +#endif #include #include @@ -46,15 +55,55 @@ #include #include #include + +#ifndef _WIN32 #include +#endif #include "mdb.h" +#include "midl.h" -#define ULONG unsigned long -typedef ULONG pgno_t; +/** @defgroup internal MDB Internals + * @{ + */ +/** @defgroup compat Windows Compatibility Macros + * @{ + */ +#ifdef _WIN32 +#define pthread_t DWORD +#define pthread_mutex_t HANDLE +#define pthread_key_t DWORD +#define pthread_self() GetCurrentThreadId() +#define pthread_key_create(x,y) *(x) = TlsAlloc() +#define pthread_key_delete(x) TlsFree(x) +#define pthread_getspecific(x) TlsGetValue(x) +#define pthread_setspecific(x,y) TlsSetValue(x,y) +#define pthread_mutex_unlock(x) ReleaseMutex(x) +#define pthread_mutex_lock(x) WaitForSingleObject(x, INFINITE) +#define LOCK_MUTEX_R(env) pthread_mutex_lock(env->me_rmutex) +#define UNLOCK_MUTEX_R(env) pthread_mutex_unlock(env->me_rmutex) +#define LOCK_MUTEX_W(env) pthread_mutex_lock(env->me_wmutex) +#define UNLOCK_MUTEX_W(env) pthread_mutex_unlock(env->me_wmutex) +#define getpid() GetCurrentProcessId() +#define fdatasync(fd) !FlushFileBuffers(fd) +#define ErrCode() GetLastError() +#define GetPageSize(x) {SYSTEM_INFO si; GetSystemInfo(&si); (x) = si.dwPageSize;} +#define close(fd) CloseHandle(fd) +#define munmap(ptr,len) UnmapViewOfFile(ptr) +#else +#define LOCK_MUTEX_R(env) pthread_mutex_lock(&env->me_txns->mti_mutex) +#define UNLOCK_MUTEX_R(env) pthread_mutex_unlock(&env->me_txns->mti_mutex) +#define LOCK_MUTEX_W(env) pthread_mutex_lock(&env->me_txns->mti_wmutex) +#define UNLOCK_MUTEX_W(env) pthread_mutex_unlock(&env->me_txns->mti_wmutex) +#define ErrCode() errno +#define HANDLE int +#define INVALID_HANDLE_VALUE -1 +#define GetPageSize(x) (x) = sysconf(_SC_PAGE_SIZE) +#endif -#include "midl.h" +/** @} */ +#ifndef _WIN32 /* Note: If O_DSYNC is undefined but exists in /usr/include, * preferably set some compiler flag to get the definition. * Otherwise compile with the less efficient -DMDB_DSYNC=O_SYNC. @@ -62,9 +111,12 @@ typedef ULONG pgno_t; #ifndef MDB_DSYNC # define MDB_DSYNC O_DSYNC #endif +#endif + +typedef ULONG pgno_t; #ifndef DEBUG -#define DEBUG 1 +#define DEBUG 0 #endif #if !(__STDC_VERSION__ >= 199901L || defined(__GNUC__)) @@ -91,6 +143,32 @@ typedef ULONG pgno_t; #define DKEY(x) #endif +/* The DB view is always consistent because all writes are wrapped in + * the wmutex. Finer-grained locks aren't necessary. + */ +#ifndef LAZY_LOCKS +#define LAZY_LOCKS 1 +#endif +#if LAZY_LOCKS +#define LAZY_MUTEX_LOCK(x) +#define LAZY_MUTEX_UNLOCK(x) +#define LAZY_RWLOCK_UNLOCK(x) +#define LAZY_RWLOCK_WRLOCK(x) +#define LAZY_RWLOCK_RDLOCK(x) +#define LAZY_RWLOCK_DEF(x) +#define LAZY_RWLOCK_INIT(x,y) +#define LAZY_RWLOCK_DESTROY(x) +#else +#define LAZY_MUTEX_LOCK(x) pthread_mutex_lock(x) +#define LAZY_MUTEX_UNLOCK(x) pthread_mutex_unlock(x) +#define LAZY_RWLOCK_UNLOCK(x) pthread_rwlock_unlock(x) +#define LAZY_RWLOCK_WRLOCK(x) pthread_rwlock_wrlock(x) +#define LAZY_RWLOCK_RDLOCK(x) pthread_rwlock_rdlock(x) +#define LAZY_RWLOCK_DEF(x) pthread_rwlock_t x +#define LAZY_RWLOCK_INIT(x,y) pthread_rwlock_init(x,y) +#define LAZY_RWLOCK_DESTROY(x) pthread_rwlock_destroy(x) +#endif + #define P_INVALID (~0UL) #define F_ISSET(w, f) (((w) & (f)) == (f)) @@ -125,7 +203,16 @@ typedef struct MDB_reader { typedef struct MDB_txbody { uint32_t mtb_magic; uint32_t mtb_version; +/* For POSIX the actual mutexes reside in shared memory. + * On Windows, mutexes are allocated by the kernel; we store + * the name in shared memory so that other processes can + * grab them. + */ +#ifdef _WIN32 + char mtb_rmname[32]; +#else pthread_mutex_t mtb_mutex; +#endif ULONG mtb_txnid; uint32_t mtb_numreaders; uint32_t mtb_me_toggle; @@ -137,14 +224,20 @@ typedef struct MDB_txninfo { #define mti_magic mt1.mtb.mtb_magic #define mti_version mt1.mtb.mtb_version #define mti_mutex mt1.mtb.mtb_mutex +#define mti_rmname mt1.mtb.mtb_rmname #define mti_txnid mt1.mtb.mtb_txnid #define mti_numreaders mt1.mtb.mtb_numreaders #define mti_me_toggle mt1.mtb.mtb_me_toggle char pad[(sizeof(MDB_txbody)+CACHELINE-1) & ~(CACHELINE-1)]; } mt1; union { +#ifdef _WIN32 + char mt2_wmname[32]; +#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)]; } mt2; MDB_reader mti_readers[1]; @@ -242,8 +335,8 @@ typedef struct MDB_pageparent { unsigned mp_pi; } MDB_pageparent; -static MDB_dpage *mdb_alloc_page(MDB_txn *txn, MDB_page *parent, unsigned int parent_idx, int num); -static int mdb_touch(MDB_txn *txn, MDB_pageparent *mp); +static MDB_dpage *mdb_alloc_page(MDB_txn *txn, MDB_dbi dbi, MDB_page *parent, unsigned int parent_idx, int num); +static int mdb_touch(MDB_txn *txn, MDB_dbi dbi, MDB_pageparent *mp); typedef struct MDB_ppage { /* ordered list of pages */ MDB_page *mp_page; @@ -257,29 +350,32 @@ typedef struct MDB_ppage { /* ordered list of pages */ struct MDB_xcursor; struct MDB_cursor { + struct MDB_xcursor *mc_xcursor; MDB_txn *mc_txn; - MDB_ppage mc_stack[CURSOR_STACK]; /* stack of parent pages */ - unsigned int mc_snum; /* number of pushed pages */ MDB_dbi mc_dbi; - short mc_initialized; /* 1 if initialized */ - short mc_eof; /* 1 if end is reached */ - struct MDB_xcursor *mc_xcursor; + unsigned int mc_snum; /* number of pushed pages */ + unsigned int mc_flags; +#define C_INITIALIZED 0x01 +#define C_EOF 0x02 +#define C_XDIRTY 0x04 + MDB_ppage mc_stack[CURSOR_STACK]; /* stack of parent pages */ }; #define METADATA(p) ((void *)((char *)(p) + PAGEHDRSZ)) +/* We guarantee 2-byte alignment for nodes */ typedef struct MDB_node { -#define mn_pgno mn_p.np_pgno -#define mn_dsize mn_p.np_dsize - union { - pgno_t np_pgno; /* child page number */ - uint32_t np_dsize; /* leaf data size */ - } mn_p; - unsigned int mn_flags:4; - unsigned int mn_ksize:12; /* key size */ + /* lo and hi are used for data size on leaf nodes and for + * child pgno on branch nodes. On 64 bit platforms, flags + * is also used for pgno. (branch nodes ignore flags) + */ + unsigned short mn_lo; + unsigned short mn_hi; + unsigned short mn_flags; #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 */ + unsigned short mn_ksize; /* key size */ char mn_data[1]; } MDB_node; @@ -295,11 +391,10 @@ typedef struct MDB_dbx { struct MDB_txn { pgno_t mt_next_pgno; /* next unallocated page */ ULONG mt_txnid; - ULONG mt_oldest; MDB_env *mt_env; pgno_t *mt_free_pgs; /* this is an IDL */ union { - MIDL2 *dirty_list; /* modified pages */ + ID2L dirty_list; /* modified pages */ MDB_reader *reader; } mt_u; MDB_dbx *mt_dbxs; /* array */ @@ -308,8 +403,8 @@ struct MDB_txn { #define MDB_TXN_RDONLY 0x01 /* read-only transaction */ #define MDB_TXN_ERROR 0x02 /* an error has occurred */ -#define MDB_TXN_METOGGLE 0x04 /* used meta page 1 */ unsigned int mt_flags; + unsigned int mt_toggle; }; /* Context for sorted-dup records */ @@ -321,9 +416,9 @@ typedef struct MDB_xcursor { } MDB_xcursor; struct MDB_env { - int me_fd; - int me_lfd; - int me_mfd; /* just for writing the meta pages */ + HANDLE me_fd; + HANDLE me_lfd; + HANDLE me_mfd; /* just for writing the meta pages */ #define MDB_FATAL_ERROR 0x80000000U uint32_t me_flags; uint32_t me_extrapad; /* unused for now */ @@ -334,7 +429,6 @@ struct MDB_env { char *me_map; MDB_txninfo *me_txns; MDB_meta *me_metas[2]; - MDB_meta *me_meta; MDB_txn *me_txn; /* current write transaction */ size_t me_mapsize; off_t me_size; /* current file size */ @@ -347,7 +441,12 @@ struct MDB_env { pthread_key_t me_txkey; /* thread-key for readers */ MDB_dpage *me_dpages; pgno_t me_free_pgs[MDB_IDL_UM_SIZE]; - MIDL2 me_dirty_list[MDB_IDL_DB_SIZE]; + ID2 me_dirty_list[MDB_IDL_UM_SIZE]; + LAZY_RWLOCK_DEF(me_dblock); +#ifdef _WIN32 + HANDLE me_rmutex; /* Windows mutexes don't reside in shared mem */ + HANDLE me_wmutex; +#endif }; #define NODESIZE offsetof(MDB_node, mn_data) @@ -357,8 +456,19 @@ struct MDB_env { #define NODEPTR(p, i) ((MDB_node *)((char *)(p) + (p)->mp_ptrs[i])) #define NODEKEY(node) (void *)((node)->mn_data) #define NODEDATA(node) (void *)((char *)(node)->mn_data + (node)->mn_ksize) -#define NODEPGNO(node) ((node)->mn_pgno) -#define NODEDSZ(node) ((node)->mn_dsize) +#if LONG_MAX == 0x7fffffff +#define NODEPGNO(node) ((node)->mn_lo | ((node)->mn_hi << 16)) +#define SETPGNO(node,pgno) do { \ + (node)->mn_lo = (pgno) & 0xffff; (node)->mn_hi = (pgno) >> 16;} while(0) +#else +#define NODEPGNO(node) ((node)->mn_lo | ((node)->mn_hi << 16) | ((unsigned long)(node)->mn_flags << 32)) +#define SETPGNO(node,pgno) do { \ + (node)->mn_lo = (pgno) & 0xffff; (node)->mn_hi = (pgno) >> 16; \ + (node)->mn_flags = (pgno) >> 32; } while(0) +#endif +#define NODEDSZ(node) ((node)->mn_lo | ((unsigned)(node)->mn_hi << 16)) +#define SETDSZ(node,size) do { \ + (node)->mn_lo = (size) & 0xffff; (node)->mn_hi = (size) >> 16;} while(0) #define NODEKSZ(node) ((node)->mn_ksize) #define LEAF2KEY(p, i, ks) ((char *)(p) + PAGEHDRSZ + ((i)*(ks))) @@ -386,97 +496,67 @@ static int mdb_add_node(MDB_txn *txn, MDB_dbi dbi, MDB_page *mp, indx_t indx, MDB_val *key, MDB_val *data, pgno_t pgno, uint8_t flags); static void mdb_del_node(MDB_page *mp, indx_t indx, int ksize); -static int mdb_del0(MDB_txn *txn, MDB_dbi dbi, unsigned int ki, +static int mdb_del0(MDB_cursor *mc, unsigned int ki, MDB_pageparent *mpp, MDB_node *leaf); -static int mdb_put0(MDB_txn *txn, MDB_dbi dbi, - MDB_val *key, MDB_val *data, unsigned int flags); static int mdb_read_data(MDB_txn *txn, MDB_node *leaf, MDB_val *data); -static int mdb_rebalance(MDB_txn *txn, MDB_dbi dbi, MDB_pageparent *mp); -static int mdb_update_key(MDB_page *mp, indx_t indx, MDB_val *key); -static int mdb_move_node(MDB_txn *txn, MDB_dbi dbi, +static int mdb_rebalance(MDB_txn *txn, MDB_dbi dbi, MDB_pageparent *mp); +static int mdb_update_key(MDB_page *mp, indx_t indx, MDB_val *key); +static int mdb_move_node(MDB_txn *txn, MDB_dbi dbi, MDB_pageparent *src, indx_t srcindx, MDB_pageparent *dst, indx_t dstindx); -static int mdb_merge(MDB_txn *txn, MDB_dbi dbi, MDB_pageparent *src, +static int mdb_merge(MDB_txn *txn, MDB_dbi dbi, MDB_pageparent *src, MDB_pageparent *dst); -static int mdb_split(MDB_txn *txn, MDB_dbi dbi, MDB_page **mpp, +static int mdb_split(MDB_txn *txn, MDB_dbi dbi, MDB_page **mpp, unsigned int *newindxp, MDB_val *newkey, MDB_val *newdata, pgno_t newpgno); static MDB_dpage *mdb_new_page(MDB_txn *txn, MDB_dbi dbi, uint32_t flags, int num); -static void cursor_pop_page(MDB_cursor *cursor); +static void cursor_pop_page(MDB_cursor *cursor); static MDB_ppage *cursor_push_page(MDB_cursor *cursor, MDB_page *mp); -static int mdb_sibling(MDB_cursor *cursor, int move_right); -static int mdb_cursor_next(MDB_cursor *cursor, +static int mdb_sibling(MDB_cursor *cursor, int move_right); +static int mdb_cursor_next(MDB_cursor *cursor, MDB_val *key, MDB_val *data, MDB_cursor_op op); -static int mdb_cursor_prev(MDB_cursor *cursor, +static int mdb_cursor_prev(MDB_cursor *cursor, MDB_val *key, MDB_val *data, MDB_cursor_op op); -static int mdb_cursor_set(MDB_cursor *cursor, +static int mdb_cursor_set(MDB_cursor *cursor, MDB_val *key, MDB_val *data, MDB_cursor_op op, int *exactp); -static int mdb_cursor_first(MDB_cursor *cursor, +static int mdb_cursor_first(MDB_cursor *cursor, MDB_val *key, MDB_val *data); -static int mdb_cursor_last(MDB_cursor *cursor, +static int mdb_cursor_last(MDB_cursor *cursor, MDB_val *key, MDB_val *data); -static void mdb_xcursor_init0(MDB_txn *txn, MDB_dbi dbi, MDB_xcursor *mx); -static void mdb_xcursor_init1(MDB_txn *txn, MDB_dbi dbi, MDB_xcursor *mx, - MDB_page *mp, MDB_node *node); -static void mdb_xcursor_fini(MDB_txn *txn, MDB_dbi dbi, MDB_xcursor *mx); +static void mdb_xcursor_init0(MDB_cursor *mc); +static void mdb_xcursor_init1(MDB_cursor *mc, MDB_page *mp, MDB_node *node); +static void mdb_xcursor_init2(MDB_cursor *mc); +static void mdb_xcursor_fini(MDB_cursor *mc); -static size_t mdb_leaf_size(MDB_env *env, MDB_val *key, +static size_t mdb_leaf_size(MDB_env *env, MDB_val *key, MDB_val *data); -static size_t mdb_branch_size(MDB_env *env, MDB_val *key); - -static int memncmp(const void *s1, size_t n1, - const void *s2, size_t n2); -static int memnrcmp(const void *s1, size_t n1, - const void *s2, size_t n2); - -static int -memncmp(const void *s1, size_t n1, const void *s2, size_t n2) -{ - int diff, len_diff = -1; - - if (n1 >= n2) { - len_diff = (n1 > n2); - n1 = n2; - } - diff = memcmp(s1, s2, n1); - return diff ? diff : len_diff; -} - -static int -memnrcmp(const void *s1, size_t n1, const void *s2, size_t n2) -{ - const unsigned char *p1, *p2, *p1_lim; +static size_t mdb_branch_size(MDB_env *env, MDB_val *key); - if (n2 == 0) - return n1 != 0; - if (n1 == 0) - return -1; +static void mdb_default_cmp(MDB_txn *txn, MDB_dbi dbi); - p1 = (const unsigned char *)s1 + n1 - 1; - p2 = (const unsigned char *)s2 + n2 - 1; +static MDB_cmp_func memncmp, memnrcmp, intcmp, cintcmp; - for (p1_lim = (n1 <= n2 ? s1 : s2); *p1 == *p2; p1--, p2--) { - if (p1 == p1_lim) - return (p1 != s1) ? (p1 != p2) : (p2 != s2) ? -1 : 0; - } - return *p1 - *p2; -} +#ifdef _WIN32 +static SECURITY_DESCRIPTOR mdb_null_sd; +static SECURITY_ATTRIBUTES mdb_all_sa; +static int mdb_sec_inited; +#endif char * mdb_version(int *maj, int *min, int *pat) { - *maj = MDB_VERSION_MAJOR; - *min = MDB_VERSION_MINOR; - *pat = MDB_VERSION_PATCH; + if (maj) *maj = MDB_VERSION_MAJOR; + if (min) *min = MDB_VERSION_MINOR; + if (pat) *pat = MDB_VERSION_PATCH; return MDB_VERSION_STRING; } -static char *const errstr[] = { +static char *const mdb_errstr[] = { "MDB_KEYEXIST: Key/data pair already exists", "MDB_NOTFOUND: No matching key/data pair found", "MDB_PAGE_NOTFOUND: Requested page not found", @@ -492,11 +572,12 @@ mdb_strerror(int err) return ("Successful return: 0"); if (err >= MDB_KEYEXIST && err <= MDB_VERSION_MISMATCH) - return errstr[err - MDB_KEYEXIST]; + return mdb_errstr[err - MDB_KEYEXIST]; return strerror(err); } +#if DEBUG static char * mdb_dkey(MDB_val *key, char *buf) { @@ -505,25 +586,20 @@ mdb_dkey(MDB_val *key, char *buf) unsigned int i; if (key->mv_size > MAXKEYSIZE) return "MAXKEYSIZE"; +#if 1 for (i=0; imv_size; i++) ptr += sprintf(ptr, "%02x", *c++); +#else + sprintf(buf, "%.*s", key->mv_size, key->mv_data); +#endif return buf; } +#endif int mdb_cmp(MDB_txn *txn, MDB_dbi dbi, const MDB_val *a, const MDB_val *b) { - if (txn->mt_dbxs[dbi].md_cmp) - return txn->mt_dbxs[dbi].md_cmp(a, b); - - if (txn->mt_dbs[dbi].md_flags & (MDB_REVERSEKEY -#if __BYTE_ORDER == __LITTLE_ENDIAN - |MDB_INTEGERKEY -#endif - )) - return memnrcmp(a->mv_data, a->mv_size, b->mv_data, b->mv_size); - else - return memncmp((char *)a->mv_data, a->mv_size, b->mv_data, b->mv_size); + return txn->mt_dbxs[dbi].md_cmp(a, b); } int @@ -531,86 +607,83 @@ mdb_dcmp(MDB_txn *txn, MDB_dbi dbi, const MDB_val *a, const MDB_val *b) { if (txn->mt_dbxs[dbi].md_dcmp) return txn->mt_dbxs[dbi].md_dcmp(a, b); - - if (txn->mt_dbs[dbi].md_flags & (0 -#if __BYTE_ORDER == __LITTLE_ENDIAN - |MDB_INTEGERDUP -#endif - )) - return memnrcmp(a->mv_data, a->mv_size, b->mv_data, b->mv_size); else - return memncmp((char *)a->mv_data, a->mv_size, b->mv_data, b->mv_size); + return EINVAL; /* too bad you can't distinguish this from a valid result */ } /* Allocate new page(s) for writing */ static MDB_dpage * -mdb_alloc_page(MDB_txn *txn, MDB_page *parent, unsigned int parent_idx, int num) +mdb_alloc_page(MDB_txn *txn, MDB_dbi dbi, MDB_page *parent, unsigned int parent_idx, int num) { MDB_dpage *dp; pgno_t pgno = P_INVALID; - ULONG oldest; - MIDL2 mid; + ID2 mid; if (txn->mt_txnid > 2) { - oldest = txn->mt_txnid - 2; - 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_pageparent mpp; - MDB_node *leaf; - ULONG *kptr; - - mpp.mp_parent = NULL; - mpp.mp_pi = 0; - mdb_search_page(txn, FREE_DBI, NULL, NULL, 0, &mpp); - leaf = NODEPTR(mpp.mp_page, 0); - kptr = (ULONG *)NODEKEY(leaf); + if (!txn->mt_env->me_pghead && dbi != FREE_DBI && + txn->mt_dbs[FREE_DBI].md_root != P_INVALID) { + /* See if there's anything in the free DB */ + MDB_cursor mc; + MDB_pageparent mpp; + MDB_node *leaf; + ULONG *kptr, oldest; - /* It's potentially usable, unless there are still - * older readers outstanding. Grab it. - */ - if (oldest > *kptr) { - MDB_oldpages *mop; - MDB_val data; - pgno_t *idl; - - mdb_read_data(txn, leaf, &data); - idl = (ULONG *)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; - txn->mt_env->me_pghead = mop; - memcpy(mop->mo_pages, idl, MDB_IDL_SIZEOF(idl)); + mpp.mp_parent = NULL; + mpp.mp_pi = 0; + mc.mc_txn = txn; + mc.mc_dbi = FREE_DBI; + mc.mc_snum = 0; + mc.mc_flags = 0; + mdb_search_page(txn, FREE_DBI, NULL, &mc, 0, &mpp); + leaf = NODEPTR(mpp.mp_page, 0); + kptr = (ULONG *)NODEKEY(leaf); -#if DEBUG > 1 { unsigned int i; - DPRINTF("IDL read txn %lu root %lu num %lu", - mop->mo_txnid, txn->mt_dbs[FREE_DBI].md_root, idl[0]); - for (i=0; imt_txnid - 1; + for (i=0; imt_env->me_txns->mti_numreaders; i++) { + ULONG mr = txn->mt_env->me_txns->mti_readers[i].mr_txnid; + if (mr && mr < oldest) + oldest = mr; } } + + if (oldest > *kptr) { + /* It's usable, grab it. + */ + MDB_oldpages *mop; + MDB_ppage *top; + MDB_val data; + pgno_t *idl; + + mdb_read_data(txn, leaf, &data); + idl = (ULONG *)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; + txn->mt_env->me_pghead = mop; + memcpy(mop->mo_pages, idl, MDB_IDL_SIZEOF(idl)); + +#if DEBUG > 1 + { + unsigned int i; + DPRINTF("IDL read txn %lu root %lu num %lu", + mop->mo_txnid, txn->mt_dbs[FREE_DBI].md_root, idl[0]); + for (i=0; imt_env->me_pghead) { - unsigned int i; - for (i=0; imt_env->me_txns->mti_numreaders; i++) { - ULONG mr = txn->mt_env->me_txns->mti_readers[i].mr_txnid; - if (!mr) continue; - if (mr < oldest) - oldest = txn->mt_env->me_txns->mti_readers[i].mr_txnid; + /* drop this IDL from the DB */ + top = CURSOR_TOP(&mc); + top->mp_ki = 0; + mc.mc_flags = C_INITIALIZED; + mdb_cursor_del(&mc, 0); + } } - if (oldest > txn->mt_env->me_pghead->mo_txnid) { + if (txn->mt_env->me_pghead) { MDB_oldpages *mop = txn->mt_env->me_pghead; - txn->mt_oldest = oldest; if (num > 1) { /* FIXME: For now, always use fresh pages. We * really ought to search the free list for a @@ -634,7 +707,6 @@ mdb_alloc_page(MDB_txn *txn, MDB_page *parent, unsigned int parent_idx, int num) } } } - } if (pgno == P_INVALID) { /* DB size is maxed out */ @@ -659,7 +731,7 @@ mdb_alloc_page(MDB_txn *txn, MDB_page *parent, unsigned int parent_idx, int num) } mid.mid = dp->p.mp_pgno; mid.mptr = dp; - mdb_midl2_insert(txn->mt_u.dirty_list, &mid); + mdb_mid2l_insert(txn->mt_u.dirty_list, &mid); return dp; } @@ -667,7 +739,7 @@ mdb_alloc_page(MDB_txn *txn, MDB_page *parent, unsigned int parent_idx, int num) /* Touch a page: make it dirty and re-insert into tree with updated pgno. */ static int -mdb_touch(MDB_txn *txn, MDB_pageparent *pp) +mdb_touch(MDB_txn *txn, MDB_dbi dbi, MDB_pageparent *pp) { MDB_page *mp = pp->mp_page; pgno_t pgno; @@ -676,11 +748,11 @@ mdb_touch(MDB_txn *txn, MDB_pageparent *pp) if (!F_ISSET(mp->mp_flags, P_DIRTY)) { MDB_dpage *dp; - if ((dp = mdb_alloc_page(txn, pp->mp_parent, pp->mp_pi, 1)) == NULL) + if ((dp = mdb_alloc_page(txn, dbi, pp->mp_parent, pp->mp_pi, 1)) == NULL) return ENOMEM; - DPRINTF("touched page %lu -> %lu", mp->mp_pgno, dp->p.mp_pgno); + DPRINTF("touched db %u page %lu -> %lu", dbi, mp->mp_pgno, dp->p.mp_pgno); assert(mp->mp_pgno != dp->p.mp_pgno); - mdb_midl_insert(txn->mt_free_pgs, mp->mp_pgno); + mdb_midl_append(txn->mt_free_pgs, mp->mp_pgno); pgno = dp->p.mp_pgno; memcpy(&dp->p, mp, txn->mt_env->me_psize); mp = &dp->p; @@ -689,7 +761,7 @@ mdb_touch(MDB_txn *txn, MDB_pageparent *pp) /* Update the page number to new touched page. */ if (pp->mp_parent != NULL) - NODEPGNO(NODEPTR(pp->mp_parent, pp->mp_pi)) = mp->mp_pgno; + SETPGNO(NODEPTR(pp->mp_parent, pp->mp_pi), mp->mp_pgno); pp->mp_page = mp; } return 0; @@ -701,7 +773,7 @@ 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)) - rc = errno; + rc = ErrCode(); } return rc; } @@ -714,70 +786,55 @@ mdb_txn_renew0(MDB_txn *txn) { MDB_env *env = txn->mt_env; - int rc, toggle; - - if (env->me_flags & MDB_FATAL_ERROR) { - DPUTS("mdb_txn_begin: environment had fatal error, must shutdown!"); - return MDB_PANIC; - } - if (txn->mt_flags & MDB_TXN_RDONLY) { MDB_reader *r = pthread_getspecific(env->me_txkey); if (!r) { unsigned int i; - pthread_mutex_lock(&env->me_txns->mti_mutex); + pid_t pid = getpid(); + pthread_t tid = pthread_self(); + + LOCK_MUTEX_R(env); for (i=0; ime_txns->mti_numreaders; i++) if (env->me_txns->mti_readers[i].mr_pid == 0) break; if (i == env->me_maxreaders) { - pthread_mutex_unlock(&env->me_txns->mti_mutex); - return ENOSPC; + UNLOCK_MUTEX_R(env); + return ENOMEM; } - env->me_txns->mti_readers[i].mr_pid = getpid(); - env->me_txns->mti_readers[i].mr_tid = pthread_self(); - r = &env->me_txns->mti_readers[i]; - pthread_setspecific(env->me_txkey, r); + env->me_txns->mti_readers[i].mr_pid = pid; + env->me_txns->mti_readers[i].mr_tid = tid; if (i >= env->me_txns->mti_numreaders) env->me_txns->mti_numreaders = i+1; - pthread_mutex_unlock(&env->me_txns->mti_mutex); + UNLOCK_MUTEX_R(env); + 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; r->mr_txnid = txn->mt_txnid; txn->mt_u.reader = r; } else { - pthread_mutex_lock(&env->me_txns->mti_wmutex); + LOCK_MUTEX_W(env); txn->mt_txnid = env->me_txns->mti_txnid+1; + 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; txn->mt_free_pgs = env->me_free_pgs; txn->mt_free_pgs[0] = 0; + txn->mt_next_pgno = env->me_metas[txn->mt_toggle]->mm_last_pg+1; env->me_txn = txn; } - toggle = env->me_txns->mti_me_toggle; - if ((rc = mdb_env_read_meta(env, &toggle)) != MDB_SUCCESS) { - mdb_txn_reset0(txn); - return rc; - } - /* Copy the DB arrays */ + LAZY_RWLOCK_RDLOCK(&env->me_dblock); txn->mt_numdbs = env->me_numdbs; txn->mt_dbxs = env->me_dbxs; /* mostly static anyway */ - memcpy(txn->mt_dbs, env->me_meta->mm_dbs, 2 * sizeof(MDB_db)); + memcpy(txn->mt_dbs, env->me_metas[txn->mt_toggle]->mm_dbs, 2 * sizeof(MDB_db)); if (txn->mt_numdbs > 2) memcpy(txn->mt_dbs+2, env->me_dbs[env->me_db_toggle]+2, (txn->mt_numdbs - 2) * sizeof(MDB_db)); - - if (!(txn->mt_flags & MDB_TXN_RDONLY)) { - if (toggle) - txn->mt_flags |= MDB_TXN_METOGGLE; - txn->mt_next_pgno = env->me_meta->mm_last_pg+1; - } - - DPRINTF("begin txn %p %lu%c on mdbenv %p, root page %lu", txn, - txn->mt_txnid, (txn->mt_flags & MDB_TXN_RDONLY) ? 'r' : 'w', - (void *) env, txn->mt_dbs[MAIN_DBI].md_root); + LAZY_RWLOCK_UNLOCK(&env->me_dblock); return MDB_SUCCESS; } @@ -790,31 +847,36 @@ mdb_txn_renew(MDB_txn *txn) if (!txn) return EINVAL; + if (txn->mt_env->me_flags & MDB_FATAL_ERROR) { + DPUTS("environment had fatal error, must shutdown!"); + return MDB_PANIC; + } + rc = mdb_txn_renew0(txn); if (rc == MDB_SUCCESS) { - DPRINTF("reset txn %p %lu%c on mdbenv %p, root page %lu", txn, - txn->mt_txnid, (txn->mt_flags & MDB_TXN_RDONLY) ? 'r' : 'w', + DPRINTF("renew txn %lu%c %p on mdbenv %p, root page %lu", + txn->mt_txnid, (txn->mt_flags & MDB_TXN_RDONLY) ? 'r' : 'w', txn, (void *)txn->mt_env, txn->mt_dbs[MAIN_DBI].md_root); } return rc; } int -mdb_txn_begin(MDB_env *env, int rdonly, MDB_txn **ret) +mdb_txn_begin(MDB_env *env, unsigned int flags, MDB_txn **ret) { MDB_txn *txn; int rc; if (env->me_flags & MDB_FATAL_ERROR) { - DPUTS("mdb_txn_begin: environment had fatal error, must shutdown!"); + DPUTS("environment had fatal error, must shutdown!"); return MDB_PANIC; } if ((txn = calloc(1, sizeof(MDB_txn) + env->me_maxdbs * sizeof(MDB_db))) == NULL) { - DPRINTF("calloc: %s", strerror(errno)); + DPRINTF("calloc: %s", strerror(ErrCode())); return ENOMEM; } txn->mt_dbs = (MDB_db *)(txn+1); - if (rdonly) { + if (flags & MDB_RDONLY) { txn->mt_flags |= MDB_TXN_RDONLY; } txn->mt_env = env; @@ -824,8 +886,8 @@ mdb_txn_begin(MDB_env *env, int rdonly, MDB_txn **ret) free(txn); else { *ret = txn; - DPRINTF("begin txn %p %lu%c on mdbenv %p, root page %lu", txn, - txn->mt_txnid, (txn->mt_flags & MDB_TXN_RDONLY) ? 'r' : 'w', + DPRINTF("begin txn %lu%c %p on mdbenv %p, root page %lu", + txn->mt_txnid, (txn->mt_flags & MDB_TXN_RDONLY) ? 'r' : 'w', txn, (void *) env, txn->mt_dbs[MAIN_DBI].md_root); } @@ -864,7 +926,7 @@ mdb_txn_reset0(MDB_txn *txn) env->me_txn = NULL; for (i=2; ime_numdbs; i++) env->me_dbxs[i].md_dirty = 0; - pthread_mutex_unlock(&env->me_txns->mti_wmutex); + UNLOCK_MUTEX_W(env); } } @@ -874,8 +936,8 @@ mdb_txn_reset(MDB_txn *txn) if (txn == NULL) return; - DPRINTF("reset txn %p %lu%c on mdbenv %p, root page %lu", txn, - txn->mt_txnid, (txn->mt_flags & MDB_TXN_RDONLY) ? 'r' : 'w', + DPRINTF("reset txn %lu%c %p on mdbenv %p, root page %lu", + txn->mt_txnid, (txn->mt_flags & MDB_TXN_RDONLY) ? 'r' : 'w', txn, (void *)txn->mt_env, txn->mt_dbs[MAIN_DBI].md_root); mdb_txn_reset0(txn); @@ -887,8 +949,8 @@ mdb_txn_abort(MDB_txn *txn) if (txn == NULL) return; - DPRINTF("abort txn %p %lu%c on mdbenv %p, root page %lu", txn, - txn->mt_txnid, (txn->mt_flags & MDB_TXN_RDONLY) ? 'r' : 'w', + DPRINTF("abort txn %lu%c %p on mdbenv %p, root page %lu", + txn->mt_txnid, (txn->mt_flags & MDB_TXN_RDONLY) ? 'r' : 'w', txn, (void *)txn->mt_env, txn->mt_dbs[MAIN_DBI].md_root); mdb_txn_reset0(txn); @@ -905,7 +967,7 @@ mdb_txn_commit(MDB_txn *txn) MDB_dpage *dp; MDB_env *env; pgno_t next; - struct iovec iov[MDB_COMMIT_PAGES]; + MDB_cursor mc; assert(txn != NULL); assert(txn->mt_env != NULL); @@ -932,35 +994,38 @@ mdb_txn_commit(MDB_txn *txn) if (!txn->mt_u.dirty_list[0].mid) goto done; - DPRINTF("committing txn %p %lu on mdbenv %p, root page %lu", txn, - txn->mt_txnid, (void *) env, txn->mt_dbs[MAIN_DBI].md_root); + DPRINTF("committing txn %lu %p on mdbenv %p, root page %lu", + txn->mt_txnid, txn, (void *)env, txn->mt_dbs[MAIN_DBI].md_root); + + mc.mc_txn = txn; + mc.mc_dbi = FREE_DBI; + mc.mc_flags = 0; /* should only be one record now */ if (env->me_pghead) { - MDB_val key, data; - MDB_oldpages *mop; + MDB_pageparent mpp; - mop = env->me_pghead; - key.mv_size = sizeof(pgno_t); - key.mv_data = (char *)&mop->mo_txnid; - data.mv_size = MDB_IDL_SIZEOF(mop->mo_pages); - data.mv_data = mop->mo_pages; - mdb_put0(txn, FREE_DBI, &key, &data, 0); - free(env->me_pghead); - env->me_pghead = NULL; + /* make sure first page of freeDB is touched and on freelist */ + mpp.mp_parent = NULL; + mpp.mp_pi = 0; + mc.mc_snum = 0; + mdb_search_page(txn, FREE_DBI, NULL, &mc, 1, &mpp); } /* save to free list */ if (!MDB_IDL_IS_ZERO(txn->mt_free_pgs)) { MDB_val key, data; MDB_pageparent mpp; + ULONG i; /* make sure last page of freeDB is touched and on freelist */ key.mv_size = MAXKEYSIZE+1; key.mv_data = NULL; mpp.mp_parent = NULL; mpp.mp_pi = 0; - mdb_search_page(txn, FREE_DBI, &key, NULL, 1, &mpp); + mc.mc_snum = 0; + mdb_search_page(txn, FREE_DBI, &key, &mc, 1, &mpp); + mdb_midl_sort(txn->mt_free_pgs); #if DEBUG > 1 { unsigned int i; @@ -975,9 +1040,34 @@ mdb_txn_commit(MDB_txn *txn) /* write to last page of freeDB */ key.mv_size = sizeof(pgno_t); key.mv_data = (char *)&txn->mt_txnid; - data.mv_size = MDB_IDL_SIZEOF(txn->mt_free_pgs); data.mv_data = txn->mt_free_pgs; - mdb_put0(txn, FREE_DBI, &key, &data, 0); + /* The free list can still grow during this call, + * despite the pre-emptive touches above. So check + * and make sure the entire thing got written. + */ + do { + i = 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]); + } + /* should only be one record now */ + if (env->me_pghead) { + MDB_val key, data; + MDB_oldpages *mop; + + mop = env->me_pghead; + key.mv_size = sizeof(pgno_t); + key.mv_data = (char *)&mop->mo_txnid; + data.mv_size = MDB_IDL_SIZEOF(mop->mo_pages); + data.mv_data = mop->mo_pages; + mdb_cursor_put(&mc, &key, &data, 0); + free(env->me_pghead); + env->me_pghead = NULL; } /* Update DB root pointers. Their pages have already been @@ -987,10 +1077,12 @@ mdb_txn_commit(MDB_txn *txn) MDB_val data; data.mv_size = sizeof(MDB_db); + mc.mc_dbi = MAIN_DBI; + mc.mc_flags = 0; for (i = 2; i < txn->mt_numdbs; i++) { if (txn->mt_dbxs[i].md_dirty) { data.mv_data = &txn->mt_dbs[i]; - mdb_put0(txn, MAIN_DBI, &txn->mt_dbxs[i].md_name, &data, 0); + mdb_cursor_put(&mc, &txn->mt_dbxs[i].md_name, &data, 0); } } } @@ -1000,6 +1092,36 @@ mdb_txn_commit(MDB_txn *txn) next = 0; i = 1; do { +#ifdef _WIN32 + /* Windows actually supports scatter/gather I/O, but only on + * unbuffered file handles. Since we're relying on the OS page + * cache for all our data, that's self-defeating. So we just + * write pages one at a time. We use the ov structure to set + * the write offset, to at least save the overhead of a Seek + * system call. + */ + OVERLAPPED ov; + memset(&ov, 0, sizeof(ov)); + for (; i<=txn->mt_u.dirty_list[0].mid; i++) { + dp = txn->mt_u.dirty_list[i].mptr; + DPRINTF("committing page %lu", dp->p.mp_pgno); + size = dp->p.mp_pgno * env->me_psize; + ov.Offset = size & 0xffffffff; + ov.OffsetHigh = size >> 16; + ov.OffsetHigh >>= 16; + /* clear dirty flag */ + dp->p.mp_flags &= ~P_DIRTY; + rc = WriteFile(env->me_fd, &dp->p, env->me_psize * dp->h.md_num, NULL, &ov); + if (!rc) { + n = ErrCode(); + DPRINTF("WriteFile: %d", n); + mdb_txn_abort(txn); + return n; + } + } + done = 1;; +#else + struct iovec iov[MDB_COMMIT_PAGES]; n = 0; done = 1; size = 0; @@ -1010,11 +1132,11 @@ mdb_txn_commit(MDB_txn *txn) DPRINTF("committing %u dirty pages", n); rc = writev(env->me_fd, iov, n); if (rc != size) { - n = errno; + n = ErrCode(); if (rc > 0) DPUTS("short write, filesystem full?"); else - DPRINTF("writev: %s", strerror(errno)); + DPRINTF("writev: %s", strerror(n)); mdb_txn_abort(txn); return n; } @@ -1044,15 +1166,15 @@ mdb_txn_commit(MDB_txn *txn) DPRINTF("committing %u dirty pages", n); rc = writev(env->me_fd, iov, n); if (rc != size) { - n = errno; + n = ErrCode(); if (rc > 0) DPUTS("short write, filesystem full?"); else - DPRINTF("writev: %s", strerror(errno)); + DPRINTF("writev: %s", strerror(n)); mdb_txn_abort(txn); return n; } - +#endif } while (!done); /* Drop the dirty pages. @@ -1075,30 +1197,32 @@ mdb_txn_commit(MDB_txn *txn) return n; } - env->me_txns->mti_txnid = txn->mt_txnid; - done: env->me_txn = NULL; /* update the DB tables */ { int toggle = !env->me_db_toggle; + MDB_db *ip, *jp; - for (i = 2; i < env->me_numdbs; i++) { - if (txn->mt_dbxs[i].md_dirty) { - env->me_dbs[toggle][i] = txn->mt_dbs[i]; - txn->mt_dbxs[i].md_dirty = 0; - } + ip = &env->me_dbs[toggle][2]; + jp = &txn->mt_dbs[2]; + LAZY_RWLOCK_WRLOCK(&env->me_dblock); + for (i = 2; i < txn->mt_numdbs; i++) { + if (ip->md_root != jp->md_root) + *ip = *jp; + ip++; jp++; } - for (i = env->me_numdbs; i < txn->mt_numdbs; i++) { - txn->mt_dbxs[i].md_dirty = 0; - env->me_dbxs[i] = txn->mt_dbxs[i]; - env->me_dbs[toggle][i] = txn->mt_dbs[i]; + + 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); } - pthread_mutex_unlock(&env->me_txns->mti_wmutex); + UNLOCK_MUTEX_W(env); free(txn); return MDB_SUCCESS; @@ -1110,20 +1234,25 @@ mdb_env_read_header(MDB_env *env, MDB_meta *meta) char page[PAGESIZE]; MDB_page *p; MDB_meta *m; - int rc; - - assert(env != NULL); + int rc, err; /* We don't know the page size yet, so use a minimum value. */ - if ((rc = pread(env->me_fd, page, PAGESIZE, 0)) == 0) { +#ifdef _WIN32 + if (!ReadFile(env->me_fd, page, PAGESIZE, (DWORD *)&rc, NULL) || rc == 0) +#else + if ((rc = read(env->me_fd, page, PAGESIZE)) == 0) +#endif + { return ENOENT; - } else if (rc != PAGESIZE) { + } + else if (rc != PAGESIZE) { + err = ErrCode(); if (rc > 0) - errno = EINVAL; - DPRINTF("read: %s", strerror(errno)); - return errno; + err = EINVAL; + DPRINTF("read: %s", strerror(err)); + return err; } p = (MDB_page *)page; @@ -1158,7 +1287,8 @@ mdb_env_init_meta(MDB_env *env, MDB_meta *meta) unsigned int psize; DPUTS("writing new meta page"); - psize = sysconf(_SC_PAGE_SIZE); + + GetPageSize(psize); meta->mm_magic = MDB_MAGIC; meta->mm_version = MDB_VERSION; @@ -1184,9 +1314,18 @@ mdb_env_init_meta(MDB_env *env, MDB_meta *meta) m = METADATA(q); memcpy(m, meta, sizeof(*meta)); +#ifdef _WIN32 + { + DWORD len; + rc = WriteFile(env->me_fd, p, psize * 2, &len, NULL); + rc = (len == psize * 2) ? MDB_SUCCESS : ErrCode(); + } +#else rc = write(env->me_fd, p, psize * 2); + rc = (rc == (int)psize * 2) ? MDB_SUCCESS : ErrCode(); +#endif free(p); - return (rc == (int)psize * 2) ? MDB_SUCCESS : errno; + return rc; } static int @@ -1197,11 +1336,14 @@ mdb_env_write_meta(MDB_txn *txn) off_t off; int rc, len, toggle; char *ptr; +#ifdef _WIN32 + OVERLAPPED ov; +#endif assert(txn != NULL); assert(txn->mt_env != NULL); - toggle = !F_ISSET(txn->mt_flags, MDB_TXN_METOGGLE); + toggle = !txn->mt_toggle; DPRINTF("writing meta page %d for root page %lu", toggle, txn->mt_dbs[MAIN_DBI].md_root); @@ -1225,10 +1367,18 @@ mdb_env_write_meta(MDB_txn *txn) off += PAGEHDRSZ; /* Write to the SYNC fd */ +#ifdef _WIN32 + { + memset(&ov, 0, sizeof(ov)); + ov.Offset = off; + WriteFile(env->me_mfd, ptr, len, (DWORD *)&rc, &ov); + } +#else rc = pwrite(env->me_mfd, ptr, len, off); +#endif if (rc != len) { int r2; - rc = errno; + rc = ErrCode(); DPUTS("write failed, disk error?"); /* On a failure, the pagecache still contains the new data. * Write some old data back, to prevent it from being used. @@ -1236,11 +1386,24 @@ mdb_env_write_meta(MDB_txn *txn) */ meta.mm_last_pg = metab.mm_last_pg; meta.mm_txnid = metab.mm_txnid; +#ifdef _WIN32 + WriteFile(env->me_fd, ptr, len, NULL, &ov); +#else r2 = pwrite(env->me_fd, ptr, len, off); +#endif env->me_flags |= MDB_FATAL_ERROR; return rc; } + /* Memory ordering issues are irrelevant; since the entire writer + * is wrapped by wmutex, all of these changes will become visible + * after the wmutex is unlocked. Since the DB is multi-version, + * readers will get consistent data regardless of how fresh or + * how stale their view of these values is. + */ + LAZY_MUTEX_LOCK(&env->me_txns->mti_mutex); txn->mt_env->me_txns->mti_me_toggle = toggle; + txn->mt_env->me_txns->mti_txnid = txn->mt_txnid; + LAZY_MUTEX_UNLOCK(&env->me_txns->mti_mutex); return MDB_SUCCESS; } @@ -1252,15 +1415,11 @@ mdb_env_read_meta(MDB_env *env, int *which) assert(env != NULL); - if (which) - toggle = *which; - else if (env->me_metas[0]->mm_txnid < env->me_metas[1]->mm_txnid) + if (env->me_metas[0]->mm_txnid < env->me_metas[1]->mm_txnid) toggle = 1; - if (env->me_meta != env->me_metas[toggle]) - env->me_meta = env->me_metas[toggle]; - DPRINTF("Using meta page %d", toggle); + *which = toggle; return MDB_SUCCESS; } @@ -1275,9 +1434,9 @@ mdb_env_create(MDB_env **env) e->me_maxreaders = DEFAULT_READERS; e->me_maxdbs = 2; - e->me_fd = -1; - e->me_lfd = -1; - e->me_mfd = -1; + e->me_fd = INVALID_HANDLE_VALUE; + e->me_lfd = INVALID_HANDLE_VALUE; + e->me_mfd = INVALID_HANDLE_VALUE; *env = e; return MDB_SUCCESS; } @@ -1294,6 +1453,8 @@ mdb_env_set_mapsize(MDB_env *env, size_t size) int mdb_env_set_maxdbs(MDB_env *env, int dbs) { + if (env->me_map) + return EINVAL; env->me_maxdbs = dbs; return MDB_SUCCESS; } @@ -1301,6 +1462,8 @@ mdb_env_set_maxdbs(MDB_env *env, int dbs) int mdb_env_set_maxreaders(MDB_env *env, int readers) { + if (env->me_map) + return EINVAL; env->me_maxreaders = readers; return MDB_SUCCESS; } @@ -1317,7 +1480,7 @@ mdb_env_get_maxreaders(MDB_env *env, int *readers) static int mdb_env_open2(MDB_env *env, unsigned int flags) { - int i, newenv = 0; + int i, newenv = 0, toggle; MDB_meta meta; MDB_page *p; @@ -1336,13 +1499,41 @@ mdb_env_open2(MDB_env *env, unsigned int flags) env->me_mapsize = newenv ? DEFAULT_MAPSIZE : meta.mm_mapsize; } +#ifdef _WIN32 + { + HANDLE mh; + LONG sizelo, sizehi; + sizelo = env->me_mapsize & 0xffffffff; + sizehi = env->me_mapsize >> 16; /* pointless on WIN32, only needed on W64 */ + sizehi >>= 16; + /* Windows won't create mappings for zero length files. + * Just allocate the maxsize right now. + */ + if (newenv) { + SetFilePointer(env->me_fd, sizelo, sizehi ? &sizehi : NULL, 0); + if (!SetEndOfFile(env->me_fd)) + return ErrCode(); + SetFilePointer(env->me_fd, 0, NULL, 0); + } + mh = CreateFileMapping(env->me_fd, NULL, PAGE_READONLY, + sizehi, sizelo, NULL); + if (!mh) + return ErrCode(); + env->me_map = MapViewOfFileEx(mh, FILE_MAP_READ, 0, 0, env->me_mapsize, + meta.mm_address); + CloseHandle(mh); + if (!env->me_map) + return ErrCode(); + } +#else i = MAP_SHARED; if (meta.mm_address && (flags & MDB_FIXEDMAP)) 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) - return errno; + return ErrCode(); +#endif if (newenv) { meta.mm_mapsize = env->me_mapsize; @@ -1362,21 +1553,23 @@ mdb_env_open2(MDB_env *env, unsigned int flags) env->me_metas[0] = METADATA(p); env->me_metas[1] = (MDB_meta *)((char *)env->me_metas[0] + meta.mm_psize); - if ((i = mdb_env_read_meta(env, NULL)) != 0) + if ((i = mdb_env_read_meta(env, &toggle)) != 0) return i; DPRINTF("opened database version %u, pagesize %u", - env->me_meta->mm_version, env->me_psize); - DPRINTF("depth: %u", env->me_meta->mm_dbs[MAIN_DBI].md_depth); - DPRINTF("entries: %lu", env->me_meta->mm_dbs[MAIN_DBI].md_entries); - DPRINTF("branch pages: %lu", env->me_meta->mm_dbs[MAIN_DBI].md_branch_pages); - DPRINTF("leaf pages: %lu", env->me_meta->mm_dbs[MAIN_DBI].md_leaf_pages); - DPRINTF("overflow pages: %lu", env->me_meta->mm_dbs[MAIN_DBI].md_overflow_pages); - DPRINTF("root: %lu", env->me_meta->mm_dbs[MAIN_DBI].md_root); + env->me_metas[toggle]->mm_version, env->me_psize); + DPRINTF("depth: %u", env->me_metas[toggle]->mm_dbs[MAIN_DBI].md_depth); + DPRINTF("entries: %lu", env->me_metas[toggle]->mm_dbs[MAIN_DBI].md_entries); + DPRINTF("branch pages: %lu", env->me_metas[toggle]->mm_dbs[MAIN_DBI].md_branch_pages); + DPRINTF("leaf pages: %lu", env->me_metas[toggle]->mm_dbs[MAIN_DBI].md_leaf_pages); + DPRINTF("overflow pages: %lu", env->me_metas[toggle]->mm_dbs[MAIN_DBI].md_overflow_pages); + DPRINTF("root: %lu", env->me_metas[toggle]->mm_dbs[MAIN_DBI].md_root); return MDB_SUCCESS; } +#ifndef _WIN32 +/* Windows doesn't support destructor callbacks for thread-specific storage */ static void mdb_env_reader_dest(void *ptr) { @@ -1386,23 +1579,41 @@ mdb_env_reader_dest(void *ptr) reader->mr_pid = 0; reader->mr_tid = 0; } +#endif /* downgrade the exclusive lock on the region back to shared */ static void mdb_env_share_locks(MDB_env *env) { - struct flock lock_info; + int toggle = 0; - env->me_txns->mti_txnid = env->me_meta->mm_txnid; if (env->me_metas[0]->mm_txnid < env->me_metas[1]->mm_txnid) - env->me_txns->mti_me_toggle = 1; + toggle = 1; + env->me_txns->mti_me_toggle = toggle; + env->me_txns->mti_txnid = env->me_metas[toggle]->mm_txnid; - memset((void *)&lock_info, 0, sizeof(lock_info)); - lock_info.l_type = F_RDLCK; - lock_info.l_whence = SEEK_SET; - lock_info.l_start = 0; - lock_info.l_len = 1; - fcntl(env->me_lfd, F_SETLK, &lock_info); +#ifdef _WIN32 + { + OVERLAPPED ov; + /* First acquire a shared lock. The Unlock will + * then release the existing exclusive lock. + */ + memset(&ov, 0, sizeof(ov)); + LockFileEx(env->me_lfd, 0, 0, 1, 0, &ov); + UnlockFile(env->me_lfd, 0, 0, 1, 0); + } +#else + { + struct flock lock_info; + /* The shared lock replaces the existing lock */ + memset((void *)&lock_info, 0, sizeof(lock_info)); + lock_info.l_type = F_RDLCK; + lock_info.l_whence = SEEK_SET; + lock_info.l_start = 0; + lock_info.l_len = 1; + fcntl(env->me_lfd, F_SETLK, &lock_info); + } +#endif } static int @@ -1410,52 +1621,138 @@ mdb_env_setup_locks(MDB_env *env, char *lpath, int mode, int *excl) { int rc; off_t size, rsize; - struct flock lock_info; *excl = 0; +#ifdef _WIN32 + if ((env->me_lfd = CreateFile(lpath, GENERIC_READ|GENERIC_WRITE, + FILE_SHARE_READ|FILE_SHARE_WRITE, NULL, OPEN_ALWAYS, + FILE_ATTRIBUTE_NORMAL, NULL)) == INVALID_HANDLE_VALUE) { + rc = ErrCode(); + return rc; + } + /* Try to get exclusive lock. If we succeed, then + * nobody is using the lock region and we should initialize it. + */ + { + if (LockFile(env->me_lfd, 0, 0, 1, 0)) { + *excl = 1; + } else { + OVERLAPPED ov; + memset(&ov, 0, sizeof(ov)); + if (!LockFileEx(env->me_lfd, 0, 0, 1, 0, &ov)) { + rc = ErrCode(); + goto fail; + } + } + } + size = GetFileSize(env->me_lfd, NULL); +#else if ((env->me_lfd = open(lpath, O_RDWR|O_CREAT, mode)) == -1) { - rc = errno; + rc = ErrCode(); return rc; } /* Try to get exclusive lock. If we succeed, then * nobody is using the lock region and we should initialize it. */ - memset((void *)&lock_info, 0, sizeof(lock_info)); - lock_info.l_type = F_WRLCK; - lock_info.l_whence = SEEK_SET; - lock_info.l_start = 0; - lock_info.l_len = 1; - rc = fcntl(env->me_lfd, F_SETLK, &lock_info); - if (rc == 0) { - *excl = 1; - } else { - lock_info.l_type = F_RDLCK; + { + struct flock lock_info; + memset((void *)&lock_info, 0, sizeof(lock_info)); + lock_info.l_type = F_WRLCK; + lock_info.l_whence = SEEK_SET; + lock_info.l_start = 0; + lock_info.l_len = 1; rc = fcntl(env->me_lfd, F_SETLK, &lock_info); - if (rc) { - rc = errno; - goto fail; + if (rc == 0) { + *excl = 1; + } else { + lock_info.l_type = F_RDLCK; + rc = fcntl(env->me_lfd, F_SETLKW, &lock_info); + if (rc) { + rc = ErrCode(); + goto fail; + } } } size = lseek(env->me_lfd, 0, SEEK_END); +#endif rsize = (env->me_maxreaders-1) * sizeof(MDB_reader) + sizeof(MDB_txninfo); if (size < rsize && *excl) { +#ifdef _WIN32 + SetFilePointer(env->me_lfd, rsize, NULL, 0); + if (!SetEndOfFile(env->me_lfd)) { + rc = ErrCode(); + goto fail; + } +#else if (ftruncate(env->me_lfd, rsize) != 0) { - rc = errno; + rc = ErrCode(); goto fail; } +#endif } else { rsize = size; size = rsize - sizeof(MDB_txninfo); env->me_maxreaders = size/sizeof(MDB_reader) + 1; } +#ifdef _WIN32 + { + HANDLE mh; + mh = CreateFileMapping(env->me_lfd, NULL, PAGE_READWRITE, + 0, 0, NULL); + if (!mh) { + rc = ErrCode(); + goto fail; + } + env->me_txns = MapViewOfFileEx(mh, FILE_MAP_WRITE, 0, 0, rsize, NULL); + CloseHandle(mh); + if (!env->me_txns) { + 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 = errno; + rc = ErrCode(); goto fail; } +#endif if (*excl) { +#ifdef _WIN32 + char *ptr; + if (!mdb_sec_inited) { + InitializeSecurityDescriptor(&mdb_null_sd, + SECURITY_DESCRIPTOR_REVISION); + SetSecurityDescriptorDacl(&mdb_null_sd, TRUE, 0, FALSE); + mdb_all_sa.nLength = sizeof(SECURITY_ATTRIBUTES); + mdb_all_sa.bInheritHandle = FALSE; + 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++ = '/'; + 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); + if (!env->me_wmutex) { + rc = ErrCode(); + goto fail; + } +#else pthread_mutexattr_t mattr; pthread_mutexattr_init(&mattr); @@ -1465,6 +1762,7 @@ 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 env->me_txns->mti_version = MDB_VERSION; env->me_txns->mti_magic = MDB_MAGIC; env->me_txns->mti_txnid = 0; @@ -1483,16 +1781,28 @@ mdb_env_setup_locks(MDB_env *env, char *lpath, int mode, int *excl) rc = MDB_VERSION_MISMATCH; goto fail; } - if (errno != EACCES && errno != EAGAIN) { - rc = errno; + rc = ErrCode(); + if (rc != EACCES && rc != EAGAIN) { + goto fail; + } +#ifdef _WIN32 + env->me_rmutex = OpenMutex(SYNCHRONIZE, FALSE, env->me_txns->mti_rmname); + if (!env->me_rmutex) { + rc = ErrCode(); + goto fail; + } + env->me_wmutex = OpenMutex(SYNCHRONIZE, FALSE, env->me_txns->mti_wmname); + if (!env->me_wmutex) { + rc = ErrCode(); goto fail; } +#endif } return MDB_SUCCESS; fail: close(env->me_lfd); - env->me_lfd = -1; + env->me_lfd = INVALID_HANDLE_VALUE; return rc; } @@ -1517,28 +1827,54 @@ mdb_env_open(MDB_env *env, const char *path, unsigned int flags, mode_t mode) if (rc) goto leave; +#ifdef _WIN32 + if (F_ISSET(flags, MDB_RDONLY)) { + oflags = GENERIC_READ; + len = OPEN_EXISTING; + } else { + oflags = GENERIC_READ|GENERIC_WRITE; + len = OPEN_ALWAYS; + } + mode = FILE_ATTRIBUTE_NORMAL; + if ((env->me_fd = CreateFile(dpath, oflags, FILE_SHARE_READ|FILE_SHARE_WRITE, + NULL, len, mode, NULL)) == INVALID_HANDLE_VALUE) { + rc = ErrCode(); + goto leave; + } +#else if (F_ISSET(flags, MDB_RDONLY)) oflags = O_RDONLY; else oflags = O_RDWR | O_CREAT; if ((env->me_fd = open(dpath, oflags, mode)) == -1) { - rc = errno; + rc = ErrCode(); goto leave; } +#endif if ((rc = mdb_env_open2(env, flags)) == MDB_SUCCESS) { /* synchronous fd for meta writes */ +#ifdef _WIN32 + if (!(flags & (MDB_RDONLY|MDB_NOSYNC))) + mode |= FILE_FLAG_WRITE_THROUGH; + if ((env->me_mfd = CreateFile(dpath, oflags, FILE_SHARE_READ|FILE_SHARE_WRITE, + NULL, len, mode, NULL)) == INVALID_HANDLE_VALUE) { + rc = ErrCode(); + goto leave; + } +#else if (!(flags & (MDB_RDONLY|MDB_NOSYNC))) oflags |= MDB_DSYNC; if ((env->me_mfd = open(dpath, oflags, mode)) == -1) { - rc = errno; + rc = ErrCode(); goto leave; } - +#endif env->me_path = strdup(path); DPRINTF("opened dbenv %p", (void *) env); pthread_key_create(&env->me_txkey, mdb_env_reader_dest); + LAZY_RWLOCK_INIT(&env->me_dblock, NULL); if (excl) mdb_env_share_locks(env); env->me_dbxs = calloc(env->me_maxdbs, sizeof(MDB_dbx)); @@ -1549,13 +1885,13 @@ mdb_env_open(MDB_env *env, const char *path, unsigned int flags, mode_t mode) leave: if (rc) { - if (env->me_fd >= 0) { + if (env->me_fd != INVALID_HANDLE_VALUE) { close(env->me_fd); - env->me_fd = -1; + env->me_fd = INVALID_HANDLE_VALUE; } - if (env->me_lfd >= 0) { + if (env->me_lfd != INVALID_HANDLE_VALUE) { close(env->me_lfd); - env->me_lfd = -1; + env->me_lfd = INVALID_HANDLE_VALUE; } } free(lpath); @@ -1581,6 +1917,7 @@ mdb_env_close(MDB_env *env) free(env->me_dbxs); free(env->me_path); + LAZY_RWLOCK_DESTROY(&env->me_dblock); pthread_key_delete(env->me_txkey); if (env->me_map) { @@ -1590,17 +1927,98 @@ mdb_env_close(MDB_env *env) close(env->me_fd); if (env->me_txns) { pid_t pid = getpid(); - size_t size = (env->me_maxreaders-1) * sizeof(MDB_reader) + sizeof(MDB_txninfo); - int i; + unsigned int i; 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, size); + munmap(env->me_txns, (env->me_maxreaders-1)*sizeof(MDB_reader)+sizeof(MDB_txninfo)); } close(env->me_lfd); free(env); } +/* only for aligned ints */ +static int +intcmp(const MDB_val *a, const MDB_val *b) +{ + if (a->mv_size == sizeof(long)) + { + unsigned long *la, *lb; + la = a->mv_data; + lb = b->mv_data; + return *la - *lb; + } else { + unsigned int *ia, *ib; + ia = a->mv_data; + ib = b->mv_data; + return *ia - *ib; + } +} + +/* ints must always be the same size */ +static int +cintcmp(const MDB_val *a, const MDB_val *b) +{ +#if __BYTE_ORDER == __LITTLE_ENDIAN + unsigned char *u, *c; + int x; + + u = a->mv_data + a->mv_size; + c = b->mv_data + a->mv_size; + while(u > (unsigned char *)a->mv_data) { + x = *--u - *--c; + if (x) break; + } + return x; +#else + return memcmp(a->mv_data, b->mv_data, a->mv_size); +#endif +} + +static int +memncmp(const MDB_val *a, const MDB_val *b) +{ + int diff, len_diff; + unsigned int len; + + len = a->mv_size; + len_diff = a->mv_size - b->mv_size; + if (len_diff > 0) + len = b->mv_size; + diff = memcmp(a->mv_data, b->mv_data, len); + return diff ? diff : len_diff; +} + +static int +memnrcmp(const MDB_val *a, const MDB_val *b) +{ + const unsigned char *p1, *p2, *p1_lim; + int diff, len_diff; + + if (b->mv_size == 0) + return a->mv_size != 0; + if (a->mv_size == 0) + return -1; + + p1 = (const unsigned char *)a->mv_data + a->mv_size - 1; + p2 = (const unsigned char *)b->mv_data + b->mv_size - 1; + + len_diff = a->mv_size - b->mv_size; + if (len_diff < 0) + p1_lim = p1 - a->mv_size; + else + p1_lim = p1 - b->mv_size; + + while (p1 > p1_lim) { + diff = *p1 - *p2; + if (diff) + return diff; + p1--; + p2--; + } + return len_diff; +} + /* Search for key within a leaf page, using binary search. * Returns the smallest entry larger or equal to the key. * If exactp is non-null, stores whether the found entry was an exact match @@ -1612,29 +2030,33 @@ static MDB_node * mdb_search_node(MDB_txn *txn, MDB_dbi dbi, MDB_page *mp, MDB_val *key, int *exactp, unsigned int *kip) { - unsigned int i = 0; + unsigned int i = 0, nkeys; int low, high; int rc = 0; MDB_node *node = NULL; MDB_val nodekey; + MDB_cmp_func *cmp; DKBUF; + nkeys = NUMKEYS(mp); + DPRINTF("searching %u keys in %s page %lu", - NUMKEYS(mp), - IS_LEAF(mp) ? "leaf" : "branch", + nkeys, IS_LEAF(mp) ? "leaf" : "branch", mp->mp_pgno); - assert(NUMKEYS(mp) > 0); - - memset(&nodekey, 0, sizeof(nodekey)); + assert(nkeys > 0); low = IS_LEAF(mp) ? 0 : 1; - high = NUMKEYS(mp) - 1; + high = nkeys - 1; + cmp = txn->mt_dbxs[dbi].md_cmp; + if (IS_LEAF2(mp)) { + nodekey.mv_size = txn->mt_dbs[dbi].md_pad; + node = NODEPTR(mp, 0); /* fake */ + } while (low <= high) { i = (low + high) >> 1; if (IS_LEAF2(mp)) { - nodekey.mv_size = txn->mt_dbs[dbi].md_pad; nodekey.mv_data = LEAF2KEY(mp, i, nodekey.mv_size); } else { node = NODEPTR(mp, i); @@ -1643,14 +2065,16 @@ mdb_search_node(MDB_txn *txn, MDB_dbi dbi, MDB_page *mp, MDB_val *key, nodekey.mv_data = NODEKEY(node); } - rc = mdb_cmp(txn, dbi, key, &nodekey); + rc = cmp(key, &nodekey); +#if DEBUG if (IS_LEAF(mp)) DPRINTF("found leaf index %u [%s], rc = %i", i, DKEY(&nodekey), rc); else DPRINTF("found branch index %u [%s -> %lu], rc = %i", i, DKEY(&nodekey), NODEPGNO(node), rc); +#endif if (rc == 0) break; @@ -1662,17 +2086,19 @@ mdb_search_node(MDB_txn *txn, MDB_dbi dbi, MDB_page *mp, MDB_val *key, if (rc > 0) { /* Found entry is less than the key. */ i++; /* Skip to get the smallest entry larger than key. */ - if (i >= NUMKEYS(mp)) - /* There is no entry larger or equal to the key. */ - return NULL; + if (!IS_LEAF2(mp)) + node = NODEPTR(mp, i); } if (exactp) *exactp = (rc == 0); if (kip) /* Store the key index if requested. */ *kip = i; + if (i >= nkeys) + /* There is no entry larger or equal to the key. */ + return NULL; /* nodeptr is fake for LEAF2 */ - return IS_LEAF2(mp) ? NODEPTR(mp, 0) : NODEPTR(mp, i); + return node; } static void @@ -1712,17 +2138,15 @@ mdb_get_page(MDB_txn *txn, pgno_t pgno, MDB_page **ret) if (!F_ISSET(txn->mt_flags, MDB_TXN_RDONLY) && txn->mt_u.dirty_list[0].mid) { MDB_dpage *dp; - MIDL2 id; unsigned x; - id.mid = pgno; - x = mdb_midl2_search(txn->mt_u.dirty_list, &id); + x = mdb_mid2l_search(txn->mt_u.dirty_list, pgno); if (x <= txn->mt_u.dirty_list[0].mid && txn->mt_u.dirty_list[x].mid == pgno) { dp = txn->mt_u.dirty_list[x].mptr; p = &dp->p; } } if (!p) { - if (pgno <= txn->mt_env->me_meta->mm_last_pg) + if (pgno <= txn->mt_env->me_metas[txn->mt_toggle]->mm_last_pg) p = (MDB_page *)(txn->mt_env->me_map + txn->mt_env->me_psize * pgno); } *ret = p; @@ -1787,8 +2211,8 @@ mdb_search_page_root(MDB_txn *txn, MDB_dbi dbi, MDB_val *key, return ENOMEM; if (modify) { - MDB_dhead *dh = ((MDB_dhead *)mp)-1; - if ((rc = mdb_touch(txn, mpp)) != 0) + MDB_dhead *dh; + if ((rc = mdb_touch(txn, dbi, mpp)) != 0) return rc; dh = ((MDB_dhead *)mpp->mp_page)-1; dh->md_parent = mpp->mp_parent; @@ -1838,7 +2262,7 @@ mdb_search_page(MDB_txn *txn, MDB_dbi dbi, MDB_val *key, return MDB_NOTFOUND; } - if (rc = mdb_get_page(txn, root, &mpp->mp_page)) + if ((rc = mdb_get_page(txn, root, &mpp->mp_page))) return rc; DPRINTF("db %u root page %lu has flags 0x%X", @@ -1857,7 +2281,7 @@ mdb_search_page(MDB_txn *txn, MDB_dbi dbi, MDB_val *key, if (!F_ISSET(mpp->mp_page->mp_flags, P_DIRTY)) { mpp->mp_parent = NULL; mpp->mp_pi = 0; - if ((rc = mdb_touch(txn, mpp))) + if ((rc = mdb_touch(txn, dbi, mpp))) return rc; txn->mt_dbs[dbi].md_root = mpp->mp_page->mp_pgno; } @@ -1874,16 +2298,16 @@ mdb_read_data(MDB_txn *txn, MDB_node *leaf, MDB_val *data) int rc; if (!F_ISSET(leaf->mn_flags, F_BIGDATA)) { - data->mv_size = leaf->mn_dsize; + data->mv_size = NODEDSZ(leaf); data->mv_data = NODEDATA(leaf); return MDB_SUCCESS; } /* Read overflow data. */ - data->mv_size = leaf->mn_dsize; + data->mv_size = NODEDSZ(leaf); memcpy(&pgno, NODEDATA(leaf), sizeof(pgno)); - if (rc = mdb_get_page(txn, pgno, &omp)) { + if ((rc = mdb_get_page(txn, pgno, &omp))) { DPRINTF("read overflow page %lu failed", pgno); return rc; } @@ -1896,9 +2320,9 @@ int mdb_get(MDB_txn *txn, MDB_dbi dbi, MDB_val *key, MDB_val *data) { - int rc, exact; - MDB_node *leaf; - MDB_pageparent mpp; + MDB_cursor mc; + MDB_xcursor mx; + int exact; DKBUF; assert(key); @@ -1912,36 +2336,16 @@ mdb_get(MDB_txn *txn, MDB_dbi dbi, return EINVAL; } - if ((rc = mdb_search_page(txn, dbi, key, NULL, 0, &mpp)) != MDB_SUCCESS) - return rc; - - leaf = mdb_search_node(txn, dbi, mpp.mp_page, key, &exact, NULL); - if (leaf && exact) { - /* Return first duplicate data item */ - if (F_ISSET(leaf->mn_flags, F_DUPDATA)) { - MDB_xcursor mx; - - mdb_xcursor_init0(txn, dbi, &mx); - mdb_xcursor_init1(txn, dbi, &mx, mpp.mp_page, leaf); - rc = mdb_search_page(&mx.mx_txn, mx.mx_cursor.mc_dbi, NULL, NULL, 0, &mpp); - if (rc != MDB_SUCCESS) - return rc; - if (IS_LEAF2(mpp.mp_page)) { - data->mv_size = txn->mt_dbs[dbi].md_pad; - data->mv_data = LEAF2KEY(mpp.mp_page, 0, data->mv_size); - } else { - leaf = NODEPTR(mpp.mp_page, 0); - data->mv_size = NODEKSZ(leaf); - data->mv_data = NODEKEY(leaf); - } - } else { - rc = mdb_read_data(txn, leaf, data); - } + mc.mc_txn = txn; + mc.mc_dbi = dbi; + mc.mc_flags = 0; + if (txn->mt_dbs[dbi].md_flags & MDB_DUPSORT) { + mc.mc_xcursor = &mx; + mdb_xcursor_init0(&mc); } else { - rc = MDB_NOTFOUND; + mc.mc_xcursor = NULL; } - - return rc; + return mdb_cursor_set(&mc, key, data, MDB_SET, &exact); } static int @@ -1979,7 +2383,7 @@ mdb_sibling(MDB_cursor *cursor, int move_right) assert(IS_BRANCH(parent->mp_page)); indx = NODEPTR(parent->mp_page, parent->mp_ki); - if (rc = mdb_get_page(cursor->mc_txn, NODEPGNO(indx), &mp)) + if ((rc = mdb_get_page(cursor->mc_txn, NODEPGNO(indx), &mp))) return rc;; #if 0 mp->parent = parent->mp_page; @@ -1999,11 +2403,11 @@ mdb_cursor_next(MDB_cursor *cursor, MDB_val *key, MDB_val *data, MDB_cursor_op o MDB_node *leaf; int rc; - if (cursor->mc_eof) { + if (cursor->mc_flags & C_EOF) { return MDB_NOTFOUND; } - assert(cursor->mc_initialized); + assert(cursor->mc_flags & C_INITIALIZED); top = CURSOR_TOP(cursor); mp = top->mp_page; @@ -2017,7 +2421,7 @@ mdb_cursor_next(MDB_cursor *cursor, MDB_val *key, MDB_val *data, MDB_cursor_op o return rc; } } else { - cursor->mc_xcursor->mx_cursor.mc_initialized = 0; + cursor->mc_xcursor->mx_cursor.mc_flags = 0; if (op == MDB_NEXT_DUP) return MDB_NOTFOUND; } @@ -2028,7 +2432,7 @@ mdb_cursor_next(MDB_cursor *cursor, MDB_val *key, MDB_val *data, MDB_cursor_op o if (top->mp_ki + 1 >= NUMKEYS(mp)) { DPUTS("=====> move to next sibling page"); if (mdb_sibling(cursor, 1) != MDB_SUCCESS) { - cursor->mc_eof = 1; + cursor->mc_flags |= C_EOF; return MDB_NOTFOUND; } top = CURSOR_TOP(cursor); @@ -2050,7 +2454,7 @@ mdb_cursor_next(MDB_cursor *cursor, MDB_val *key, MDB_val *data, MDB_cursor_op o leaf = NODEPTR(mp, top->mp_ki); if (F_ISSET(leaf->mn_flags, F_DUPDATA)) { - mdb_xcursor_init1(cursor->mc_txn, cursor->mc_dbi, cursor->mc_xcursor, mp, leaf); + mdb_xcursor_init1(cursor, mp, leaf); } if (data) { if ((rc = mdb_read_data(cursor->mc_txn, leaf, data) != MDB_SUCCESS)) @@ -2075,7 +2479,7 @@ mdb_cursor_prev(MDB_cursor *cursor, MDB_val *key, MDB_val *data, MDB_cursor_op o MDB_node *leaf; int rc; - assert(cursor->mc_initialized); + assert(cursor->mc_flags & C_INITIALIZED); top = CURSOR_TOP(cursor); mp = top->mp_page; @@ -2088,7 +2492,7 @@ mdb_cursor_prev(MDB_cursor *cursor, MDB_val *key, MDB_val *data, MDB_cursor_op o if (op != MDB_PREV || rc == MDB_SUCCESS) return rc; } else { - cursor->mc_xcursor->mx_cursor.mc_initialized = 0; + cursor->mc_xcursor->mx_cursor.mc_flags = 0; if (op == MDB_PREV_DUP) return MDB_NOTFOUND; } @@ -2100,7 +2504,7 @@ mdb_cursor_prev(MDB_cursor *cursor, MDB_val *key, MDB_val *data, MDB_cursor_op o if (top->mp_ki == 0) { DPUTS("=====> move to prev sibling page"); if (mdb_sibling(cursor, 0) != MDB_SUCCESS) { - cursor->mc_initialized = 0; + cursor->mc_flags &= ~C_INITIALIZED; return MDB_NOTFOUND; } top = CURSOR_TOP(cursor); @@ -2110,7 +2514,7 @@ mdb_cursor_prev(MDB_cursor *cursor, MDB_val *key, MDB_val *data, MDB_cursor_op o } else top->mp_ki--; - cursor->mc_eof = 0; + cursor->mc_flags &= ~C_EOF; DPRINTF("==> cursor points to page %lu with %u keys, key index %u", mp->mp_pgno, NUMKEYS(mp), top->mp_ki); @@ -2125,7 +2529,7 @@ mdb_cursor_prev(MDB_cursor *cursor, MDB_val *key, MDB_val *data, MDB_cursor_op o leaf = NODEPTR(mp, top->mp_ki); if (F_ISSET(leaf->mn_flags, F_DUPDATA)) { - mdb_xcursor_init1(cursor->mc_txn, cursor->mc_dbi, cursor->mc_xcursor, mp, leaf); + mdb_xcursor_init1(cursor, mp, leaf); } if (data) { if ((rc = mdb_read_data(cursor->mc_txn, leaf, data) != MDB_SUCCESS)) @@ -2157,27 +2561,69 @@ mdb_cursor_set(MDB_cursor *cursor, MDB_val *key, MDB_val *data, assert(key->mv_size > 0); /* See if we're already on the right page */ - if (cursor->mc_initialized) { + if (cursor->mc_flags & C_INITIALIZED) { MDB_val nodekey; + top = CURSOR_TOP(cursor); - /* Don't try this for LEAF2 pages. Maybe support that later. */ - if ((top->mp_page->mp_flags & (P_LEAF|P_LEAF2)) == P_LEAF) { + if (top->mp_page->mp_flags & P_LEAF2) { + nodekey.mv_size = cursor->mc_txn->mt_dbs[cursor->mc_dbi].md_pad; + nodekey.mv_data = LEAF2KEY(top->mp_page, 0, nodekey.mv_size); + } else { leaf = NODEPTR(top->mp_page, 0); MDB_SET_KEY(leaf, &nodekey); - rc = mdb_cmp(cursor->mc_txn, cursor->mc_dbi, key, &nodekey); - if (rc >= 0) { - leaf = NODEPTR(top->mp_page, NUMKEYS(top->mp_page)-1); - MDB_SET_KEY(leaf, &nodekey); - rc = mdb_cmp(cursor->mc_txn, cursor->mc_dbi, key, &nodekey); - if (rc <= 0) { - /* we're already on the right page */ + } + rc = cursor->mc_txn->mt_dbxs[cursor->mc_dbi].md_cmp(key, &nodekey); + if (rc == 0) { + /* Probably happens rarely, but first node on the page + * was the one we wanted. + */ + top->mp_ki = 0; +set1: + mpp.mp_page = top->mp_page; + if (exactp) + *exactp = 1; + rc = 0; + leaf = NODEPTR(top->mp_page, top->mp_ki); + goto set3; + } + if (rc > 0) { + unsigned int i; + if (NUMKEYS(top->mp_page) > 1) { + if (top->mp_page->mp_flags & P_LEAF2) { + nodekey.mv_data = LEAF2KEY(top->mp_page, + NUMKEYS(top->mp_page)-1, nodekey.mv_size); + } else { + leaf = NODEPTR(top->mp_page, NUMKEYS(top->mp_page)-1); + MDB_SET_KEY(leaf, &nodekey); + } + rc = cursor->mc_txn->mt_dbxs[cursor->mc_dbi].md_cmp(key, &nodekey); + if (rc == 0) { + /* last node was the one we wanted */ + top->mp_ki = NUMKEYS(top->mp_page)-1; + goto set1; + } + if (rc < 0) { + /* This is definitely the right page, skip search_page */ mpp.mp_page = top->mp_page; rc = 0; goto set2; } } + /* If any parents have right-sibs, search. + * Otherwise, there's nothing further. + */ + for (i=0; imc_snum-1; i++) + if (cursor->mc_stack[i].mp_ki < + NUMKEYS(cursor->mc_stack[i].mp_page)-1) + break; + if (i == cursor->mc_snum - 1) { + /* There are no other pages */ + top->mp_ki = NUMKEYS(top->mp_page); + return MDB_NOTFOUND; + } } } + cursor->mc_snum = 0; rc = mdb_search_page(cursor->mc_txn, cursor->mc_dbi, key, cursor, 0, &mpp); @@ -2205,8 +2651,9 @@ set2: leaf = NODEPTR(mpp.mp_page, 0); } - cursor->mc_initialized = 1; - cursor->mc_eof = 0; +set3: + cursor->mc_flags |= C_INITIALIZED; + cursor->mc_flags &= ~C_EOF; if (IS_LEAF2(mpp.mp_page)) { key->mv_size = cursor->mc_txn->mt_dbs[cursor->mc_dbi].md_pad; @@ -2215,7 +2662,7 @@ set2: } if (F_ISSET(leaf->mn_flags, F_DUPDATA)) { - mdb_xcursor_init1(cursor->mc_txn, cursor->mc_dbi, cursor->mc_xcursor, mpp.mp_page, leaf); + mdb_xcursor_init1(cursor, mpp.mp_page, leaf); } if (data) { if (F_ISSET(leaf->mn_flags, F_DUPDATA)) { @@ -2225,6 +2672,7 @@ set2: int ex2, *ex2p; if (op == MDB_GET_BOTH) { ex2p = &ex2; + ex2 = 0; } else { ex2p = NULL; } @@ -2236,7 +2684,7 @@ set2: MDB_val d2; if ((rc = mdb_read_data(cursor->mc_txn, leaf, &d2)) != MDB_SUCCESS) return rc; - rc = mdb_dcmp(cursor->mc_txn, cursor->mc_dbi, data, &d2); + rc = cursor->mc_txn->mt_dbxs[cursor->mc_dbi].md_dcmp(data, &d2); if (rc) { if (op == MDB_GET_BOTH || rc > 0) return MDB_NOTFOUND; @@ -2271,8 +2719,8 @@ mdb_cursor_first(MDB_cursor *cursor, MDB_val *key, MDB_val *data) assert(IS_LEAF(mpp.mp_page)); leaf = NODEPTR(mpp.mp_page, 0); - cursor->mc_initialized = 1; - cursor->mc_eof = 0; + cursor->mc_flags |= C_INITIALIZED; + cursor->mc_flags &= ~C_EOF; if (IS_LEAF2(mpp.mp_page)) { key->mv_size = cursor->mc_txn->mt_dbs[cursor->mc_dbi].md_pad; @@ -2282,13 +2730,13 @@ mdb_cursor_first(MDB_cursor *cursor, MDB_val *key, MDB_val *data) if (data) { if (F_ISSET(leaf->mn_flags, F_DUPDATA)) { - mdb_xcursor_init1(cursor->mc_txn, cursor->mc_dbi, cursor->mc_xcursor, mpp.mp_page, leaf); + mdb_xcursor_init1(cursor, mpp.mp_page, leaf); rc = mdb_cursor_first(&cursor->mc_xcursor->mx_cursor, data, NULL); if (rc) return rc; } else { if (cursor->mc_xcursor) - cursor->mc_xcursor->mx_cursor.mc_initialized = 0; + cursor->mc_xcursor->mx_cursor.mc_flags = 0; if ((rc = mdb_read_data(cursor->mc_txn, leaf, data)) != MDB_SUCCESS) return rc; } @@ -2317,8 +2765,8 @@ mdb_cursor_last(MDB_cursor *cursor, MDB_val *key, MDB_val *data) assert(IS_LEAF(mpp.mp_page)); leaf = NODEPTR(mpp.mp_page, NUMKEYS(mpp.mp_page)-1); - cursor->mc_initialized = 1; - cursor->mc_eof = 0; + cursor->mc_flags |= C_INITIALIZED; + cursor->mc_flags &= ~C_EOF; top = CURSOR_TOP(cursor); top->mp_ki = NUMKEYS(top->mp_page) - 1; @@ -2331,7 +2779,7 @@ mdb_cursor_last(MDB_cursor *cursor, MDB_val *key, MDB_val *data) if (data) { if (F_ISSET(leaf->mn_flags, F_DUPDATA)) { - mdb_xcursor_init1(cursor->mc_txn, cursor->mc_dbi, cursor->mc_xcursor, mpp.mp_page, leaf); + mdb_xcursor_init1(cursor, mpp.mp_page, leaf); rc = mdb_cursor_last(&cursor->mc_xcursor->mx_cursor, data, NULL); if (rc) return rc; @@ -2374,12 +2822,13 @@ mdb_cursor_get(MDB_cursor *cursor, MDB_val *key, MDB_val *data, case MDB_GET_MULTIPLE: if (data == NULL || !(cursor->mc_txn->mt_dbs[cursor->mc_dbi].md_flags & MDB_DUPFIXED) || - !cursor->mc_initialized) { + !(cursor->mc_flags & C_INITIALIZED)) { rc = EINVAL; break; } rc = MDB_SUCCESS; - if (!cursor->mc_xcursor->mx_cursor.mc_initialized || cursor->mc_xcursor->mx_cursor.mc_eof) + if (!(cursor->mc_xcursor->mx_cursor.mc_flags & C_INITIALIZED) || + (cursor->mc_xcursor->mx_cursor.mc_flags & C_EOF)) break; goto fetchm; case MDB_NEXT_MULTIPLE: @@ -2388,12 +2837,12 @@ mdb_cursor_get(MDB_cursor *cursor, MDB_val *key, MDB_val *data, rc = EINVAL; break; } - if (!cursor->mc_initialized) + if (!(cursor->mc_flags & C_INITIALIZED)) rc = mdb_cursor_first(cursor, key, data); else rc = mdb_cursor_next(cursor, key, data, MDB_NEXT_DUP); if (rc == MDB_SUCCESS) { - if (cursor->mc_xcursor->mx_cursor.mc_initialized) { + if (cursor->mc_xcursor->mx_cursor.mc_flags & C_INITIALIZED) { MDB_ppage *top; fetchm: top = CURSOR_TOP(&cursor->mc_xcursor->mx_cursor); @@ -2409,7 +2858,7 @@ fetchm: case MDB_NEXT: case MDB_NEXT_DUP: case MDB_NEXT_NODUP: - if (!cursor->mc_initialized) + if (!(cursor->mc_flags & C_INITIALIZED)) rc = mdb_cursor_first(cursor, key, data); else rc = mdb_cursor_next(cursor, key, data, op); @@ -2417,7 +2866,7 @@ fetchm: case MDB_PREV: case MDB_PREV_DUP: case MDB_PREV_NODUP: - if (!cursor->mc_initialized || cursor->mc_eof) + if (!(cursor->mc_flags & C_INITIALIZED) || (cursor->mc_flags & C_EOF)) rc = mdb_cursor_last(cursor, key, data); else rc = mdb_cursor_prev(cursor, key, data, op); @@ -2425,9 +2874,29 @@ fetchm: case MDB_FIRST: rc = mdb_cursor_first(cursor, key, data); break; + case MDB_FIRST_DUP: + if (data == NULL || + !(cursor->mc_txn->mt_dbs[cursor->mc_dbi].md_flags & MDB_DUPSORT) || + !(cursor->mc_flags & C_INITIALIZED) || + !(cursor->mc_xcursor->mx_cursor.mc_flags & C_INITIALIZED)) { + rc = EINVAL; + break; + } + rc = mdb_cursor_first(&cursor->mc_xcursor->mx_cursor, data, NULL); + break; case MDB_LAST: rc = mdb_cursor_last(cursor, key, data); break; + case MDB_LAST_DUP: + if (data == NULL || + !(cursor->mc_txn->mt_dbs[cursor->mc_dbi].md_flags & MDB_DUPSORT) || + !(cursor->mc_flags & C_INITIALIZED) || + !(cursor->mc_xcursor->mx_cursor.mc_flags & C_INITIALIZED)) { + rc = EINVAL; + break; + } + rc = mdb_cursor_last(&cursor->mc_xcursor->mx_cursor, data, NULL); + break; default: DPRINTF("unhandled/unimplemented cursor operation %u", op); rc = EINVAL; @@ -2437,6 +2906,337 @@ fetchm: return rc; } +static int +mdb_cursor_touch(MDB_cursor *mc) +{ + MDB_pageparent mpp; + MDB_dhead *dh; + unsigned int i; + int rc; + + mpp.mp_parent = NULL; + mpp.mp_pi = 0; + if (mc->mc_dbi > MAIN_DBI && !mc->mc_txn->mt_dbxs[mc->mc_dbi].md_dirty) { + rc = mdb_search_page(mc->mc_txn, MAIN_DBI, &mc->mc_txn->mt_dbxs[mc->mc_dbi].md_name, + NULL, 1, &mpp); + if (rc) return rc; + mc->mc_txn->mt_dbxs[mc->mc_dbi].md_dirty = 1; + } + for(i=0; imc_snum; i++) { + mpp.mp_page = mc->mc_stack[i].mp_page; + if (!F_ISSET(mpp.mp_page->mp_flags, P_DIRTY)) { + rc = mdb_touch(mc->mc_txn, mc->mc_dbi, &mpp); + if (rc) return rc; + mc->mc_stack[i].mp_page = mpp.mp_page; + if (!i) { + mc->mc_txn->mt_dbs[mc->mc_dbi].md_root = mpp.mp_page->mp_pgno; + } + } + dh = ((MDB_dhead *)mpp.mp_page)-1; + dh->md_parent = mpp.mp_parent; + dh->md_pi = mpp.mp_pi; + mpp.mp_parent = mpp.mp_page; + mpp.mp_pi = mc->mc_stack[i].mp_ki; + } + return MDB_SUCCESS; +} + +int +mdb_cursor_put(MDB_cursor *mc, MDB_val *key, MDB_val *data, + unsigned int flags) +{ + MDB_ppage *top; + MDB_node *leaf; + MDB_val xdata, *rdata, dkey; + MDB_db dummy; + char dbuf[PAGESIZE]; + int do_sub = 0; + size_t nsize; + DKBUF; + int rc, rc2; + + 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); + + dkey.mv_size = 0; + + if (flags == MDB_CURRENT) { + if (!(mc->mc_flags & C_INITIALIZED)) + return EINVAL; + rc = MDB_SUCCESS; + } else if (mc->mc_txn->mt_dbs[mc->mc_dbi].md_root == P_INVALID) { + MDB_dpage *dp; + /* new database, write a root leaf page */ + DPUTS("allocating new root leaf page"); + if ((dp = mdb_new_page(mc->mc_txn, mc->mc_dbi, P_LEAF, 1)) == NULL) { + return ENOMEM; + } + mc->mc_snum = 0; + cursor_push_page(mc, &dp->p); + mc->mc_txn->mt_dbs[mc->mc_dbi].md_root = dp->p.mp_pgno; + mc->mc_txn->mt_dbs[mc->mc_dbi].md_depth++; + mc->mc_txn->mt_dbxs[mc->mc_dbi].md_dirty = 1; + if ((mc->mc_txn->mt_dbs[mc->mc_dbi].md_flags & (MDB_DUPSORT|MDB_DUPFIXED)) + == MDB_DUPFIXED) + dp->p.mp_flags |= P_LEAF2; + mc->mc_flags |= C_INITIALIZED; + rc = MDB_NOTFOUND; + goto top; + } else { + int exact = 0; + rc = mdb_cursor_set(mc, key, NULL, MDB_SET, &exact); + if (flags == MDB_NOOVERWRITE && rc == 0) { + DPRINTF("duplicate key [%s]", DKEY(key)); + return MDB_KEYEXIST; + } + if (rc && rc != MDB_NOTFOUND) + return rc; + } + + /* Cursor is positioned, now make sure all pages are writable */ + rc2 = mdb_cursor_touch(mc); + if (rc2) return rc2; + +top: + top = CURSOR_TOP(mc); + + /* The key already exists */ + if (rc == MDB_SUCCESS) { + /* there's only a key anyway, so this is a no-op */ + if (IS_LEAF2(top->mp_page)) { + unsigned int ksize = mc->mc_txn->mt_dbs[mc->mc_dbi].md_pad; + if (key->mv_size != ksize) + return EINVAL; + if (flags == MDB_CURRENT) { + char *ptr = LEAF2KEY(top->mp_page, top->mp_ki, ksize); + memcpy(ptr, key->mv_data, ksize); + } + return MDB_SUCCESS; + } + + leaf = NODEPTR(top->mp_page, top->mp_ki); + + /* DB has dups? */ + if (F_ISSET(mc->mc_txn->mt_dbs[mc->mc_dbi].md_flags, MDB_DUPSORT)) { + /* Was a single item before, must convert now */ + if (!F_ISSET(leaf->mn_flags, F_DUPDATA)) { + dkey.mv_size = NODEDSZ(leaf); + dkey.mv_data = dbuf; + memcpy(dbuf, NODEDATA(leaf), dkey.mv_size); + /* data matches, ignore it */ + if (!mdb_dcmp(mc->mc_txn, mc->mc_dbi, data, &dkey)) + return (flags == MDB_NODUPDATA) ? MDB_KEYEXIST : MDB_SUCCESS; + memset(&dummy, 0, sizeof(dummy)); + if (mc->mc_txn->mt_dbs[mc->mc_dbi].md_flags & MDB_DUPFIXED) { + dummy.md_pad = data->mv_size; + dummy.md_flags = MDB_DUPFIXED; + if (mc->mc_txn->mt_dbs[mc->mc_dbi].md_flags & MDB_INTEGERDUP) + dummy.md_flags |= MDB_INTEGERKEY; + } + dummy.md_root = P_INVALID; + if (dkey.mv_size == sizeof(MDB_db)) { + memcpy(NODEDATA(leaf), &dummy, sizeof(dummy)); + goto put_sub; + } + mdb_del_node(top->mp_page, top->mp_ki, 0); + do_sub = 1; + rdata = &xdata; + xdata.mv_size = sizeof(MDB_db); + xdata.mv_data = &dummy; + goto new_sub; + } + goto put_sub; + } + /* 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); + goto done; + } + mdb_del_node(top->mp_page, top->mp_ki, 0); + } else { + DPRINTF("inserting key at index %i", top->mp_ki); + } + + rdata = data; + +new_sub: + nsize = IS_LEAF2(top->mp_page) ? key->mv_size : mdb_leaf_size(mc->mc_txn->mt_env, key, rdata); + if (SIZELEFT(top->mp_page) < nsize) { + rc = mdb_split(mc->mc_txn, mc->mc_dbi, &top->mp_page, &top->mp_ki, key, rdata, P_INVALID); + } else { + /* There is room already in this leaf page. */ + rc = mdb_add_node(mc->mc_txn, mc->mc_dbi, top->mp_page, top->mp_ki, key, rdata, 0, 0); + } + + 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(top->mp_page, top->mp_ki); + 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) { + leaf = NODEPTR(top->mp_page, top->mp_ki); +put_sub: + if (flags == MDB_CURRENT) + mdb_xcursor_init2(mc); + else + mdb_xcursor_init1(mc, top->mp_page, leaf); + xdata.mv_size = 0; + xdata.mv_data = ""; + if (flags == MDB_NODUPDATA) + flags = MDB_NOOVERWRITE; + /* converted, write the original data first */ + if (dkey.mv_size) { + rc = mdb_cursor_put(&mc->mc_xcursor->mx_cursor, &dkey, &xdata, flags); + if (rc) return rc; + leaf->mn_flags |= F_DUPDATA; + } + rc = mdb_cursor_put(&mc->mc_xcursor->mx_cursor, data, &xdata, flags); + mdb_xcursor_fini(mc); + memcpy(NODEDATA(leaf), + &mc->mc_xcursor->mx_txn.mt_dbs[mc->mc_xcursor->mx_cursor.mc_dbi], + sizeof(MDB_db)); + } + mc->mc_txn->mt_dbs[mc->mc_dbi].md_entries++; + } +done: + return rc; +} + +int +mdb_cursor_del(MDB_cursor *mc, unsigned int flags) +{ + MDB_pageparent mpp; + MDB_ppage *top; + MDB_node *leaf; + int rc; + + if (F_ISSET(mc->mc_txn->mt_flags, MDB_TXN_RDONLY)) + return EACCES; + + if (!mc->mc_flags & C_INITIALIZED) + return EINVAL; + + rc = mdb_cursor_touch(mc); + if (rc) return rc; + + top = CURSOR_TOP(mc); + leaf = NODEPTR(top->mp_page, top->mp_ki); + mpp.mp_page = top->mp_page; + if (mc->mc_snum > 1) { + MDB_ppage *parent = CURSOR_PARENT(mc); + mpp.mp_parent = parent->mp_page; + mpp.mp_pi = parent->mp_ki; + } else { + mpp.mp_parent = NULL; + mpp.mp_pi = 0; + } + + if (!IS_LEAF2(top->mp_page) && F_ISSET(leaf->mn_flags, F_DUPDATA)) { + MDB_pageparent mp2; + + if (flags != MDB_NODUPDATA) { + mdb_xcursor_init2(mc); + rc = mdb_cursor_del(&mc->mc_xcursor->mx_cursor, 0); + mdb_xcursor_fini(mc); + /* If sub-DB still has entries, we're done */ + if (mc->mc_xcursor->mx_txn.mt_dbs[mc->mc_xcursor->mx_cursor.mc_dbi].md_root + != P_INVALID) { + memcpy(NODEDATA(leaf), + &mc->mc_xcursor->mx_txn.mt_dbs[mc->mc_xcursor->mx_cursor.mc_dbi], + sizeof(MDB_db)); + mc->mc_txn->mt_dbs[mc->mc_dbi].md_entries--; + return rc; + } + /* otherwise fall thru and delete the sub-DB */ + } + + /* add all the child DB's pages to the free list */ + mc->mc_xcursor->mx_cursor.mc_snum = 0; + rc = mdb_search_page(&mc->mc_xcursor->mx_txn, mc->mc_xcursor->mx_cursor.mc_dbi, + NULL, &mc->mc_xcursor->mx_cursor, 0, &mp2); + if (rc == MDB_SUCCESS) { + MDB_ppage *top, *parent; + MDB_node *ni; + unsigned int i; +#if 0 + MDB_dpage *dp; + ID2 mid; + int dirty_root = 0; +#endif + + mc->mc_txn->mt_dbs[mc->mc_dbi].md_entries -= + mc->mc_xcursor->mx_txn.mt_dbs[mc->mc_xcursor->mx_cursor.mc_dbi].md_entries; + + cursor_pop_page(&mc->mc_xcursor->mx_cursor); + if (mc->mc_xcursor->mx_cursor.mc_snum) { +#if 0 + if (mc->mc_xcursor->mx_cursor.mc_stack[0].mp_page->mp_flags & P_DIRTY) { + dirty_root = 1; + } +#endif + while (mc->mc_xcursor->mx_cursor.mc_snum > 1) { + top = CURSOR_TOP(&mc->mc_xcursor->mx_cursor); + parent = CURSOR_PARENT(&mc->mc_xcursor->mx_cursor); + for (i=0; imp_page); i++) { + MDB_page *mp; + pgno_t pg; + ni = NODEPTR(top->mp_page, i); + pg = NODEPGNO(ni); + if ((rc = mdb_get_page(mc->mc_txn, pg, &mp))) + return rc; +#if 0 + if (mp->mp_flags & P_DIRTY) { + /* drop it */ + mid.mid = pg; + mdb_mid2l_delete(mc->mc_txn->mt_u.dirty_list, &mid); + dp = mid.mptr; + dp->h.md_parent = (MDB_page *)mc->mc_txn->mt_env->me_dpages; + mc->mc_txn->mt_env->me_dpages = dp; + } else +#endif + { + /* free it */ + mdb_midl_append(mc->mc_txn->mt_free_pgs, pg); + } + } + rc = mdb_sibling(&mc->mc_xcursor->mx_cursor, 1); + if (rc) break; + } + } +#if 0 + if (dirty_root) { + /* drop it */ + mid.mid = mc->mc_xcursor->mx_txn.mt_dbs[mc->mc_xcursor->mx_cursor.mc_dbi].md_root; + mdb_mid2l_delete(mc->mc_txn->mt_u.dirty_list, &mid); + dp = mid.mptr; + dp->h.md_parent = (MDB_page *)mc->mc_txn->mt_env->me_dpages; + mc->mc_txn->mt_env->me_dpages = dp; + } else +#endif + { + /* free it */ + mdb_midl_append(mc->mc_txn->mt_free_pgs, + mc->mc_xcursor->mx_txn.mt_dbs[mc->mc_xcursor->mx_cursor.mc_dbi].md_root); + } + } + } + + return mdb_del0(mc, top->mp_ki, &mpp, leaf); +} + /* Allocate a page and initialize it */ static MDB_dpage * @@ -2444,7 +3244,7 @@ mdb_new_page(MDB_txn *txn, MDB_dbi dbi, uint32_t flags, int num) { MDB_dpage *dp; - if ((dp = mdb_alloc_page(txn, NULL, 0, num)) == NULL) + if ((dp = mdb_alloc_page(txn, dbi, NULL, 0, num)) == NULL) return NULL; DPRINTF("allocated new mpage %lu, page size %u", dp->p.mp_pgno, txn->mt_env->me_psize); @@ -2474,6 +3274,7 @@ mdb_leaf_size(MDB_env *env, MDB_val *key, MDB_val *data) /* put on overflow page */ sz -= data->mv_size - sizeof(pgno_t); } + sz += sz & 1; return sz + sizeof(indx_t); } @@ -2506,10 +3307,10 @@ mdb_add_node(MDB_txn *txn, MDB_dbi dbi, MDB_page *mp, indx_t indx, assert(mp->mp_upper >= mp->mp_lower); - DPRINTF("add node [%s] to %s page %lu at index %i, key size %zu", - key ? DKEY(key) : NULL, + DPRINTF("add to %s page %lu index %i, data size %zu key size %zu [%s]", IS_LEAF(mp) ? "leaf" : "branch", - mp->mp_pgno, indx, key ? key->mv_size : 0); + mp->mp_pgno, indx, data ? data->mv_size : 0, + key ? key->mv_size : 0, key ? DKEY(key) : NULL); if (IS_LEAF2(mp)) { /* Move higher keys up one slot. */ @@ -2549,6 +3350,7 @@ mdb_add_node(MDB_txn *txn, MDB_dbi dbi, MDB_page *mp, indx_t indx, node_size += data->mv_size; } } + node_size += node_size & 1; if (node_size + sizeof(indx_t) > SIZELEFT(mp)) { DPRINTF("not enough room in page %lu, got %u ptrs", @@ -2575,9 +3377,9 @@ mdb_add_node(MDB_txn *txn, MDB_dbi dbi, MDB_page *mp, indx_t indx, node->mn_ksize = (key == NULL) ? 0 : key->mv_size; node->mn_flags = flags; if (IS_LEAF(mp)) - node->mn_dsize = data->mv_size; + SETDSZ(node,data->mv_size); else - NODEPGNO(node) = pgno; + SETPGNO(node,pgno); if (key) memcpy(NODEKEY(node), key->mv_data, key->mv_size); @@ -2631,6 +3433,7 @@ mdb_del_node(MDB_page *mp, indx_t indx, int ksize) else sz += NODEDSZ(node); } + sz += sz & 1; ptr = mp->mp_ptrs[indx]; numkeys = NUMKEYS(mp); @@ -2651,17 +3454,18 @@ mdb_del_node(MDB_page *mp, indx_t indx, int ksize) } static void -mdb_xcursor_init0(MDB_txn *txn, MDB_dbi dbi, MDB_xcursor *mx) +mdb_xcursor_init0(MDB_cursor *mc) { + MDB_xcursor *mx = mc->mc_xcursor; MDB_dbi dbn; - mx->mx_txn = *txn; + mx->mx_txn = *mc->mc_txn; mx->mx_txn.mt_dbxs = mx->mx_dbxs; mx->mx_txn.mt_dbs = mx->mx_dbs; - mx->mx_dbxs[0] = txn->mt_dbxs[0]; - mx->mx_dbxs[1] = txn->mt_dbxs[1]; - if (dbi > 1) { - mx->mx_dbxs[2] = txn->mt_dbxs[dbi]; + mx->mx_dbxs[0] = mc->mc_txn->mt_dbxs[0]; + mx->mx_dbxs[1] = mc->mc_txn->mt_dbxs[1]; + if (mc->mc_dbi > 1) { + mx->mx_dbxs[2] = mc->mc_txn->mt_dbxs[mc->mc_dbi]; dbn = 2; } else { dbn = 1; @@ -2672,50 +3476,67 @@ mdb_xcursor_init0(MDB_txn *txn, MDB_dbi dbi, MDB_xcursor *mx) mx->mx_dbxs[dbn+1].md_dirty = 0; mx->mx_txn.mt_numdbs = dbn+2; - mx->mx_cursor.mc_snum = 0; + mx->mx_cursor.mc_xcursor = NULL; mx->mx_cursor.mc_txn = &mx->mx_txn; mx->mx_cursor.mc_dbi = dbn+1; } static void -mdb_xcursor_init1(MDB_txn *txn, MDB_dbi dbi, MDB_xcursor *mx, MDB_page *mp, MDB_node *node) +mdb_xcursor_init1(MDB_cursor *mc, MDB_page *mp, MDB_node *node) { MDB_db *db = NODEDATA(node); + MDB_xcursor *mx = mc->mc_xcursor; MDB_dbi dbn; - mx->mx_dbs[0] = txn->mt_dbs[0]; - mx->mx_dbs[1] = txn->mt_dbs[1]; - if (dbi > 1) { - mx->mx_dbs[2] = txn->mt_dbs[dbi]; + mx->mx_dbs[0] = mc->mc_txn->mt_dbs[0]; + mx->mx_dbs[1] = mc->mc_txn->mt_dbs[1]; + if (mc->mc_dbi > 1) { + mx->mx_dbs[2] = mc->mc_txn->mt_dbs[mc->mc_dbi]; + mx->mx_dbxs[2].md_dirty = mc->mc_txn->mt_dbxs[mc->mc_dbi].md_dirty; dbn = 3; } else { dbn = 2; } - DPRINTF("Sub-db %u for db %u root page %lu", dbn, dbi, db->md_root); + DPRINTF("Sub-db %u for db %u root page %lu", dbn, mc->mc_dbi, db->md_root); mx->mx_dbs[dbn] = *db; if (F_ISSET(mp->mp_flags, P_DIRTY)) mx->mx_dbxs[dbn].md_dirty = 1; mx->mx_dbxs[dbn].md_name.mv_data = NODEKEY(node); mx->mx_dbxs[dbn].md_name.mv_size = node->mn_ksize; - mx->mx_txn.mt_next_pgno = txn->mt_next_pgno; - mx->mx_txn.mt_oldest = txn->mt_oldest; - mx->mx_txn.mt_u = txn->mt_u; - mx->mx_cursor.mc_initialized = 0; - mx->mx_cursor.mc_eof = 0; + mx->mx_txn.mt_next_pgno = mc->mc_txn->mt_next_pgno; + mx->mx_txn.mt_u = mc->mc_txn->mt_u; + mx->mx_cursor.mc_snum = 0; + mx->mx_cursor.mc_flags = 0; } static void -mdb_xcursor_fini(MDB_txn *txn, MDB_dbi dbi, MDB_xcursor *mx) +mdb_xcursor_init2(MDB_cursor *mc) { - txn->mt_next_pgno = mx->mx_txn.mt_next_pgno; - txn->mt_oldest = mx->mx_txn.mt_oldest; - txn->mt_u = mx->mx_txn.mt_u; - txn->mt_dbs[0] = mx->mx_dbs[0]; - txn->mt_dbs[1] = mx->mx_dbs[1]; - txn->mt_dbxs[0].md_dirty = mx->mx_dbxs[0].md_dirty; - txn->mt_dbxs[1].md_dirty = mx->mx_dbxs[1].md_dirty; - if (dbi > 1) { - txn->mt_dbs[dbi] = mx->mx_dbs[2]; - txn->mt_dbxs[dbi].md_dirty = mx->mx_dbxs[2].md_dirty; + MDB_xcursor *mx = mc->mc_xcursor; + MDB_dbi dbn; + mx->mx_dbs[0] = mc->mc_txn->mt_dbs[0]; + mx->mx_dbs[1] = mc->mc_txn->mt_dbs[1]; + if (mc->mc_dbi > 1) { + mx->mx_dbs[2] = mc->mc_txn->mt_dbs[mc->mc_dbi]; + mx->mx_dbxs[2].md_dirty = mc->mc_txn->mt_dbxs[mc->mc_dbi].md_dirty; + dbn = 3; + } else { + dbn = 2; + } + DPRINTF("Sub-db %u for db %u root page %lu", dbn, mc->mc_dbi, + mx->mx_dbs[dbn].md_root); + mx->mx_txn.mt_next_pgno = mc->mc_txn->mt_next_pgno; +} + +static void +mdb_xcursor_fini(MDB_cursor *mc) +{ + MDB_xcursor *mx = mc->mc_xcursor; + mc->mc_txn->mt_next_pgno = mx->mx_txn.mt_next_pgno; + mc->mc_txn->mt_dbs[0] = mx->mx_dbs[0]; + mc->mc_txn->mt_dbs[1] = mx->mx_dbs[1]; + if (mc->mc_dbi > 1) { + mc->mc_txn->mt_dbs[mc->mc_dbi] = mx->mx_dbs[2]; + mc->mc_txn->mt_dbxs[mc->mc_dbi].md_dirty = mx->mx_dbxs[2].md_dirty; } } @@ -2737,7 +3558,7 @@ mdb_cursor_open(MDB_txn *txn, MDB_dbi dbi, MDB_cursor **ret) if (txn->mt_dbs[dbi].md_flags & MDB_DUPSORT) { MDB_xcursor *mx = (MDB_xcursor *)(cursor + 1); cursor->mc_xcursor = mx; - mdb_xcursor_init0(txn, dbi, mx); + mdb_xcursor_init0(cursor); } } else { return ENOMEM; @@ -2766,7 +3587,7 @@ mdb_cursor_count(MDB_cursor *mc, unsigned long *countp) if (!F_ISSET(leaf->mn_flags, F_DUPDATA)) { *countp = 1; } else { - if (!mc->mc_xcursor->mx_cursor.mc_initialized) + if (!(mc->mc_xcursor->mx_cursor.mc_flags & C_INITIALIZED)) return EINVAL; *countp = mc->mc_xcursor->mx_txn.mt_dbs[mc->mc_xcursor->mx_cursor.mc_dbi].md_entries; @@ -2774,6 +3595,32 @@ mdb_cursor_count(MDB_cursor *mc, unsigned long *countp) return MDB_SUCCESS; } +#if 0 +static void +mdb_cursor_reset(MDB_cursor *mc) +{ + if (mc != NULL) { + if (mc->mc_flags & C_XDIRTY) { + + mdb_xcursor_fini(mc); + + /* If sub-DB still has entries, update root record */ + if (mc->mc_xcursor->mx_txn.mt_dbs[mc->mc_xcursor->mx_cursor.mc_dbi].md_root + != P_INVALID) { + MDB_ppage *top; + MDB_node *leaf; + top = CURSOR_TOP(mc); + leaf = NODEPTR(top->mp_page, top->mp_ki); + memcpy(NODEDATA(leaf), + &mc->mc_xcursor->mx_txn.mt_dbs[mc->mc_xcursor->mx_cursor.mc_dbi], + sizeof(MDB_db)); + } + mc->mc_flags ^= C_XDIRTY; + } + } +} +#endif + void mdb_cursor_close(MDB_cursor *cursor) { @@ -2839,8 +3686,8 @@ mdb_move_node(MDB_txn *txn, MDB_dbi dbi, MDB_pageparent *src, indx_t srcindx, DKBUF; /* Mark src and dst as dirty. */ - if ((rc = mdb_touch(txn, src)) || - (rc = mdb_touch(txn, dst))) + if ((rc = mdb_touch(txn, dbi, src)) || + (rc = mdb_touch(txn, dbi, dst))) return rc;; if (IS_LEAF2(src->mp_page)) { @@ -2933,8 +3780,8 @@ mdb_merge(MDB_txn *txn, MDB_dbi dbi, MDB_pageparent *src, MDB_pageparent *dst) assert(dst->mp_parent); /* Mark src and dst as dirty. */ - if ((rc = mdb_touch(txn, src)) || - (rc = mdb_touch(txn, dst))) + if ((rc = mdb_touch(txn, dbi, src)) || + (rc = mdb_touch(txn, dbi, dst))) return rc; /* Move all nodes from src to dst. @@ -3020,10 +3867,12 @@ mdb_rebalance(MDB_txn *txn, MDB_dbi dbi, MDB_pageparent *mpp) txn->mt_dbs[dbi].md_root = P_INVALID; txn->mt_dbs[dbi].md_depth = 0; txn->mt_dbs[dbi].md_leaf_pages = 0; + mdb_midl_append(txn->mt_free_pgs, mpp->mp_page->mp_pgno); } else if (IS_BRANCH(mpp->mp_page) && NUMKEYS(mpp->mp_page) == 1) { DPUTS("collapsing root page!"); + mdb_midl_append(txn->mt_free_pgs, mpp->mp_page->mp_pgno); txn->mt_dbs[dbi].md_root = NODEPGNO(NODEPTR(mpp->mp_page, 0)); - if (rc = mdb_get_page(txn, txn->mt_dbs[dbi].md_root, &root)) + if ((rc = mdb_get_page(txn, txn->mt_dbs[dbi].md_root, &root))) return rc; txn->mt_dbs[dbi].md_depth--; txn->mt_dbs[dbi].md_branch_pages--; @@ -3049,7 +3898,7 @@ mdb_rebalance(MDB_txn *txn, MDB_dbi dbi, MDB_pageparent *mpp) */ DPUTS("reading right neighbor"); node = NODEPTR(mpp->mp_parent, mpp->mp_pi + 1); - if (rc = mdb_get_page(txn, NODEPGNO(node), &npp.mp_page)) + if ((rc = mdb_get_page(txn, NODEPGNO(node), &npp.mp_page))) return rc; npp.mp_pi = mpp->mp_pi + 1; si = 0; @@ -3059,7 +3908,7 @@ mdb_rebalance(MDB_txn *txn, MDB_dbi dbi, MDB_pageparent *mpp) */ DPUTS("reading left neighbor"); node = NODEPTR(mpp->mp_parent, mpp->mp_pi - 1); - if (rc = mdb_get_page(txn, NODEPGNO(node), &npp.mp_page)) + if ((rc = mdb_get_page(txn, NODEPGNO(node), &npp.mp_page))) return rc; npp.mp_pi = mpp->mp_pi - 1; si = NUMKEYS(npp.mp_page) - 1; @@ -3086,7 +3935,7 @@ mdb_rebalance(MDB_txn *txn, MDB_dbi dbi, MDB_pageparent *mpp) } static int -mdb_del0(MDB_txn *txn, MDB_dbi dbi, unsigned int ki, MDB_pageparent *mpp, MDB_node *leaf) +mdb_del0(MDB_cursor *mc, unsigned int ki, MDB_pageparent *mpp, MDB_node *leaf) { int rc; @@ -3096,31 +3945,31 @@ mdb_del0(MDB_txn *txn, MDB_dbi dbi, unsigned int ki, MDB_pageparent *mpp, MDB_no pgno_t pg; memcpy(&pg, NODEDATA(leaf), sizeof(pg)); - ovpages = OVPAGES(NODEDSZ(leaf), txn->mt_env->me_psize); + ovpages = OVPAGES(NODEDSZ(leaf), mc->mc_txn->mt_env->me_psize); for (i=0; imt_free_pgs, pg); + mdb_midl_append(mc->mc_txn->mt_free_pgs, pg); pg++; } } - mdb_del_node(mpp->mp_page, ki, txn->mt_dbs[dbi].md_pad); - txn->mt_dbs[dbi].md_entries--; - rc = mdb_rebalance(txn, dbi, mpp); + mdb_del_node(mpp->mp_page, ki, mc->mc_txn->mt_dbs[mc->mc_dbi].md_pad); + mc->mc_txn->mt_dbs[mc->mc_dbi].md_entries--; + rc = mdb_rebalance(mc->mc_txn, mc->mc_dbi, mpp); if (rc != MDB_SUCCESS) - txn->mt_flags |= MDB_TXN_ERROR; + mc->mc_txn->mt_flags |= MDB_TXN_ERROR; return rc; } int mdb_del(MDB_txn *txn, MDB_dbi dbi, - MDB_val *key, MDB_val *data, - unsigned int flags) + MDB_val *key, MDB_val *data) { + MDB_cursor mc; + MDB_xcursor mx; + MDB_cursor_op op; + MDB_val rdata, *xdata; int rc, exact; - unsigned int ki; - MDB_node *leaf; - MDB_pageparent mpp; DKBUF; assert(key != NULL); @@ -3131,77 +3980,36 @@ mdb_del(MDB_txn *txn, MDB_dbi dbi, return EINVAL; if (F_ISSET(txn->mt_flags, MDB_TXN_RDONLY)) { - return EINVAL; + return EACCES; } if (key->mv_size == 0 || key->mv_size > MAXKEYSIZE) { return EINVAL; } - mpp.mp_parent = NULL; - mpp.mp_pi = 0; - if ((rc = mdb_search_page(txn, dbi, key, NULL, 1, &mpp)) != MDB_SUCCESS) - return rc; - - leaf = mdb_search_node(txn, dbi, mpp.mp_page, key, &exact, &ki); - if (leaf == NULL || !exact) { - return MDB_NOTFOUND; + mc.mc_txn = txn; + mc.mc_dbi = dbi; + mc.mc_flags = 0; + if (txn->mt_dbs[dbi].md_flags & MDB_DUPSORT) { + mc.mc_xcursor = &mx; + mdb_xcursor_init0(&mc); + } else { + mc.mc_xcursor = NULL; } - if (!IS_LEAF2(mpp.mp_page) && F_ISSET(leaf->mn_flags, F_DUPDATA)) { - MDB_xcursor mx; - MDB_pageparent mp2; - - mdb_xcursor_init0(txn, dbi, &mx); - mdb_xcursor_init1(txn, dbi, &mx, mpp.mp_page, leaf); - if (flags == MDB_DEL_DUP) { - rc = mdb_del(&mx.mx_txn, mx.mx_cursor.mc_dbi, data, NULL, 0); - mdb_xcursor_fini(txn, dbi, &mx); - /* If sub-DB still has entries, we're done */ - if (mx.mx_txn.mt_dbs[mx.mx_cursor.mc_dbi].md_root != P_INVALID) { - memcpy(NODEDATA(leaf), &mx.mx_txn.mt_dbs[mx.mx_cursor.mc_dbi], - sizeof(MDB_db)); - txn->mt_dbs[dbi].md_entries--; - return rc; - } - /* otherwise fall thru and delete the sub-DB */ - } else { - /* add all the child DB's pages to the free list */ - rc = mdb_search_page(&mx.mx_txn, mx.mx_cursor.mc_dbi, - NULL, &mx.mx_cursor, 0, &mp2); - if (rc == MDB_SUCCESS) { - MDB_ppage *top, *parent; - MDB_node *ni; - unsigned int i; - - cursor_pop_page(&mx.mx_cursor); - if (mx.mx_cursor.mc_snum) { - top = CURSOR_TOP(&mx.mx_cursor); - while (mx.mx_cursor.mc_snum > 1) { - parent = CURSOR_PARENT(&mx.mx_cursor); - for (i=0; imp_page); i++) { - ni = NODEPTR(top->mp_page, i); - mdb_midl_insert(txn->mt_free_pgs, NODEPGNO(ni)); - } - parent->mp_ki++; - if (parent->mp_ki >= NUMKEYS(parent->mp_page)) { - cursor_pop_page(&mx.mx_cursor); - top = parent; - } else { - ni = NODEPTR(parent->mp_page, parent->mp_ki); - rc = mdb_get_page(&mx.mx_txn, NODEPGNO(ni), &top->mp_page); - } - } - } - mdb_midl_insert(txn->mt_free_pgs, mx.mx_txn.mt_dbs[mx.mx_cursor.mc_dbi].md_root); - } - } + exact = 0; + if (data) { + op = MDB_GET_BOTH; + rdata = *data; + xdata = &rdata; + } else { + op = MDB_SET; + xdata = NULL; } - - if (data && (rc = mdb_read_data(txn, leaf, data)) != MDB_SUCCESS) - return rc; - - return mdb_del0(txn, dbi, ki, &mpp, leaf); + rc = mdb_cursor_set(&mc, key, xdata, op, &exact); + if (rc == 0) + rc = mdb_cursor_del(&mc, data ? 0 : MDB_NODUPDATA); + return rc; } /* Split page <*mpp>, and insert in either left or @@ -3217,10 +4025,10 @@ mdb_split(MDB_txn *txn, MDB_dbi dbi, MDB_page **mpp, unsigned int *newindxp, int rc = MDB_SUCCESS, ins_new = 0; indx_t newindx; pgno_t pgno = 0; - unsigned int i, j, split_indx; + unsigned int i, j, split_indx, nkeys, pmax; MDB_node *node; MDB_val sepkey, rkey, rdata; - MDB_page *copy; + MDB_page *copy, *cptr; MDB_dpage *mdp, *rdp, *pdp; MDB_dhead *dh; DKBUF; @@ -3259,12 +4067,13 @@ mdb_split(MDB_txn *txn, MDB_dbi dbi, MDB_page **mpp, unsigned int *newindxp, rdp->h.md_pi = mdp->h.md_pi + 1; DPRINTF("new right sibling: page %lu", rdp->p.mp_pgno); - split_indx = NUMKEYS(&mdp->p) / 2 + 1; + nkeys = NUMKEYS(&mdp->p); + split_indx = nkeys / 2 + 1; if (IS_LEAF2(&rdp->p)) { char *split, *ins; int x; - unsigned int nkeys = NUMKEYS(&mdp->p), lsize, rsize, ksize; + unsigned int lsize, rsize, ksize; /* Move half of the keys to the right sibling */ copy = NULL; x = *newindxp - split_indx; @@ -3304,22 +4113,54 @@ mdb_split(MDB_txn *txn, MDB_dbi dbi, MDB_page **mpp, unsigned int *newindxp, goto newsep; } - /* Move half of the keys to the right sibling. */ - if ((copy = malloc(txn->mt_env->me_psize)) == NULL) - return ENOMEM; - memcpy(copy, &mdp->p, txn->mt_env->me_psize); - memset(&mdp->p.mp_ptrs, 0, txn->mt_env->me_psize - PAGEHDRSZ); - mdp->p.mp_lower = PAGEHDRSZ; - mdp->p.mp_upper = txn->mt_env->me_psize; + /* For leaf pages, check the split point based on what + * fits where, since otherwise add_node can fail. + */ + if (IS_LEAF(&mdp->p)) { + unsigned int psize, nsize; + /* Maximum free space in an empty page */ + pmax = txn->mt_env->me_psize - PAGEHDRSZ; + nsize = mdb_leaf_size(txn->mt_env, newkey, newdata); + if (newindx < split_indx) { + psize = nsize; + for (i=0; ip, i); + psize += NODESIZE + NODEKSZ(node) + sizeof(indx_t); + if (F_ISSET(node->mn_flags, F_BIGDATA)) + psize += sizeof(pgno_t); + else + psize += NODEDSZ(node); + psize += psize & 1; + if (psize > pmax) { + split_indx = i; + break; + } + } + } else { + psize = nsize; + for (i=nkeys-1; i>=split_indx; i--) { + node = NODEPTR(&mdp->p, i); + psize += NODESIZE + NODEKSZ(node) + sizeof(indx_t); + if (F_ISSET(node->mn_flags, F_BIGDATA)) + psize += sizeof(pgno_t); + else + psize += NODEDSZ(node); + psize += psize & 1; + if (psize > pmax) { + split_indx = i+1; + break; + } + } + } + } /* First find the separating key between the split pages. */ - memset(&sepkey, 0, sizeof(sepkey)); if (newindx == split_indx) { sepkey.mv_size = newkey->mv_size; sepkey.mv_data = newkey->mv_data; } else { - node = NODEPTR(copy, split_indx); + node = NODEPTR(&mdp->p, split_indx); sepkey.mv_size = node->mn_ksize; sepkey.mv_data = NODEKEY(node); } @@ -3349,20 +4190,24 @@ newsep: return rc; } if (rc != MDB_SUCCESS) { - free(copy); return rc; } - for (i = j = 0; i <= NUMKEYS(copy); j++) { - if (i < split_indx) { - /* Re-insert in left sibling. */ - pdp = mdp; - } else { - /* Insert in right sibling. */ - if (i == split_indx) - /* Reset insert index for right sibling. */ - j = (i == newindx && ins_new); - pdp = rdp; + /* Move half of the keys to the right sibling. */ + if ((copy = malloc(txn->mt_env->me_psize)) == NULL) + return ENOMEM; + + copy->mp_pgno = mdp->p.mp_pgno; + copy->mp_flags = mdp->p.mp_flags; + copy->mp_lower = PAGEHDRSZ; + copy->mp_upper = txn->mt_env->me_psize; + cptr = copy; + for (i = j = 0; i <= nkeys; j++) { + if (i == split_indx) { + /* Insert in right sibling. */ + /* Reset insert index for right sibling. */ + j = (i == newindx && ins_new); + cptr = &rdp->p; } if (i == newindx && !ins_new) { @@ -3380,16 +4225,17 @@ newsep: /* Update page and index for the new key. */ *newindxp = j; - *mpp = &pdp->p; - } else if (i == NUMKEYS(copy)) { + if (cptr == &rdp->p) + *mpp = cptr; + } else if (i == nkeys) { break; } else { - node = NODEPTR(copy, i); + node = NODEPTR(&mdp->p, i); rkey.mv_data = NODEKEY(node); rkey.mv_size = node->mn_ksize; if (IS_LEAF(&mdp->p)) { rdata.mv_data = NODEDATA(node); - rdata.mv_size = node->mn_dsize; + rdata.mv_size = NODEDSZ(node); } else pgno = NODEPGNO(node); flags = node->mn_flags; @@ -3402,167 +4248,27 @@ newsep: rkey.mv_size = 0; } - rc = mdb_add_node(txn, dbi, &pdp->p, j, &rkey, &rdata, pgno,flags); + rc = mdb_add_node(txn, dbi, cptr, j, &rkey, &rdata, pgno, flags); } + nkeys = NUMKEYS(copy); + for (i=0; ip.mp_ptrs[i] = copy->mp_ptrs[i]; + mdp->p.mp_lower = copy->mp_lower; + mdp->p.mp_upper = copy->mp_upper; + memcpy(NODEPTR(&mdp->p, nkeys-1), NODEPTR(copy, nkeys-1), + txn->mt_env->me_psize - copy->mp_upper); free(copy); return rc; } -static int -mdb_put0(MDB_txn *txn, MDB_dbi dbi, - MDB_val *key, MDB_val *data, unsigned int flags) -{ - int rc = MDB_SUCCESS, exact; - unsigned int ki; - MDB_node *leaf; - MDB_pageparent mpp; - MDB_val xdata, *rdata, dkey; - MDB_db dummy; - char dbuf[PAGESIZE]; - int do_sub = 0; - size_t nsize; - DKBUF; - - DPRINTF("==> put db %u key [%s], size %zu, data size %zu", - dbi, DKEY(key), key->mv_size, data->mv_size); - - dkey.mv_size = 0; - mpp.mp_parent = NULL; - mpp.mp_pi = 0; - rc = mdb_search_page(txn, dbi, key, NULL, 1, &mpp); - if (rc == MDB_SUCCESS) { - leaf = mdb_search_node(txn, dbi, mpp.mp_page, key, &exact, &ki); - if (leaf && exact) { - if (flags == MDB_NOOVERWRITE) { - DPRINTF("duplicate key [%s]", DKEY(key)); - return MDB_KEYEXIST; - } - /* there's only a key anyway, so this is a no-op */ - if (IS_LEAF2(mpp.mp_page)) - return MDB_SUCCESS; - - if (F_ISSET(txn->mt_dbs[dbi].md_flags, MDB_DUPSORT)) { - /* Was a single item before, must convert now */ - if (!F_ISSET(leaf->mn_flags, F_DUPDATA)) { - dkey.mv_size = NODEDSZ(leaf); - dkey.mv_data = dbuf; - memcpy(dbuf, NODEDATA(leaf), dkey.mv_size); - /* data matches, ignore it */ - if (!mdb_dcmp(txn, dbi, data, &dkey)) - return (flags == MDB_NODUPDATA) ? MDB_KEYEXIST : MDB_SUCCESS; - memset(&dummy, 0, sizeof(dummy)); - if (txn->mt_dbs[dbi].md_flags & MDB_DUPFIXED) { - dummy.md_pad = data->mv_size; - dummy.md_flags = MDB_DUPFIXED; - if (txn->mt_dbs[dbi].md_flags & MDB_INTEGERDUP) - dummy.md_flags |= MDB_INTEGERKEY; - } - dummy.md_root = P_INVALID; - if (dkey.mv_size == sizeof(MDB_db)) { - memcpy(NODEDATA(leaf), &dummy, sizeof(dummy)); - goto put_sub; - } - mdb_del_node(mpp.mp_page, ki, 0); - do_sub = 1; - rdata = &xdata; - xdata.mv_size = sizeof(MDB_db); - xdata.mv_data = &dummy; - goto new_sub; - } - goto put_sub; - } - /* same size, just replace it */ - if (NODEDSZ(leaf) == data->mv_size) { - memcpy(NODEDATA(leaf), data->mv_data, data->mv_size); - goto done; - } - mdb_del_node(mpp.mp_page, ki, 0); - } - if (leaf == NULL) { /* append if not found */ - ki = NUMKEYS(mpp.mp_page); - DPRINTF("appending key at index %i", ki); - } - } else if (rc == MDB_NOTFOUND) { - MDB_dpage *dp; - /* new file, just write a root leaf page */ - DPUTS("allocating new root leaf page"); - if ((dp = mdb_new_page(txn, dbi, P_LEAF, 1)) == NULL) { - return ENOMEM; - } - mpp.mp_page = &dp->p; - txn->mt_dbs[dbi].md_root = mpp.mp_page->mp_pgno; - txn->mt_dbs[dbi].md_depth++; - txn->mt_dbxs[dbi].md_dirty = 1; - if ((txn->mt_dbs[dbi].md_flags & (MDB_DUPSORT|MDB_DUPFIXED)) == MDB_DUPFIXED) - mpp.mp_page->mp_flags |= P_LEAF2; - ki = 0; - } - else - goto done; - - assert(IS_LEAF(mpp.mp_page)); - DPRINTF("there are %u keys, should insert new key at index %i", - NUMKEYS(mpp.mp_page), ki); - - rdata = data; - -new_sub: - nsize = IS_LEAF2(mpp.mp_page) ? key->mv_size : mdb_leaf_size(txn->mt_env, key, rdata); - if (SIZELEFT(mpp.mp_page) < nsize) { - rc = mdb_split(txn, dbi, &mpp.mp_page, &ki, key, rdata, P_INVALID); - } else { - /* There is room already in this leaf page. */ - rc = mdb_add_node(txn, dbi, mpp.mp_page, ki, key, rdata, 0, 0); - } - - if (rc != MDB_SUCCESS) - txn->mt_flags |= MDB_TXN_ERROR; - else { - /* Remember if we just added a subdatabase */ - if (flags & F_SUBDATA) { - leaf = NODEPTR(mpp.mp_page, ki); - 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_xcursor mx; - - leaf = NODEPTR(mpp.mp_page, ki); -put_sub: - mdb_xcursor_init0(txn, dbi, &mx); - mdb_xcursor_init1(txn, dbi, &mx, mpp.mp_page, leaf); - xdata.mv_size = 0; - xdata.mv_data = ""; - if (flags == MDB_NODUPDATA) - flags = MDB_NOOVERWRITE; - /* converted, write the original data first */ - if (dkey.mv_size) { - rc = mdb_put0(&mx.mx_txn, mx.mx_cursor.mc_dbi, &dkey, &xdata, flags); - if (rc) return rc; - leaf->mn_flags |= F_DUPDATA; - } - rc = mdb_put0(&mx.mx_txn, mx.mx_cursor.mc_dbi, data, &xdata, flags); - mdb_xcursor_fini(txn, dbi, &mx); - memcpy(NODEDATA(leaf), &mx.mx_txn.mt_dbs[mx.mx_cursor.mc_dbi], - sizeof(MDB_db)); - } - txn->mt_dbs[dbi].md_entries++; - } - -done: - return rc; -} - int mdb_put(MDB_txn *txn, MDB_dbi dbi, MDB_val *key, MDB_val *data, unsigned int flags) { + MDB_cursor mc; + MDB_xcursor mx; + assert(key != NULL); assert(data != NULL); @@ -3570,7 +4276,7 @@ mdb_put(MDB_txn *txn, MDB_dbi dbi, return EINVAL; if (F_ISSET(txn->mt_flags, MDB_TXN_RDONLY)) { - return EINVAL; + return EACCES; } if (key->mv_size == 0 || key->mv_size > MAXKEYSIZE) { @@ -3580,7 +4286,17 @@ mdb_put(MDB_txn *txn, MDB_dbi dbi, if ((flags & (MDB_NOOVERWRITE|MDB_NODUPDATA)) != flags) return EINVAL; - return mdb_put0(txn, dbi, key, data, flags); + mc.mc_txn = txn; + mc.mc_dbi = dbi; + mc.mc_snum = 0; + mc.mc_flags = 0; + if (txn->mt_dbs[dbi].md_flags & MDB_DUPSORT) { + mc.mc_xcursor = &mx; + mdb_xcursor_init0(&mc); + } else { + mc.mc_xcursor = NULL; + } + return mdb_cursor_put(&mc, key, data, flags); } int @@ -3631,10 +4347,40 @@ mdb_stat0(MDB_env *env, MDB_db *db, MDB_stat *arg) int mdb_env_stat(MDB_env *env, MDB_stat *arg) { + int toggle; + if (env == NULL || arg == NULL) return EINVAL; - return mdb_stat0(env, &env->me_meta->mm_dbs[MAIN_DBI], arg); + mdb_env_read_meta(env, &toggle); + + return mdb_stat0(env, &env->me_metas[toggle]->mm_dbs[MAIN_DBI], arg); +} + +static void +mdb_default_cmp(MDB_txn *txn, MDB_dbi dbi) +{ + if (txn->mt_dbs[dbi].md_flags & MDB_REVERSEKEY) + txn->mt_dbxs[dbi].md_cmp = memnrcmp; + else if (txn->mt_dbs[dbi].md_flags & MDB_INTEGERKEY) + txn->mt_dbxs[dbi].md_cmp = cintcmp; + else + txn->mt_dbxs[dbi].md_cmp = memncmp; + + if (txn->mt_dbs[dbi].md_flags & MDB_DUPSORT) { + if (txn->mt_dbs[dbi].md_flags & MDB_INTEGERDUP) { + if (txn->mt_dbs[dbi].md_flags & MDB_DUPFIXED) + txn->mt_dbxs[dbi].md_dcmp = intcmp; + else + txn->mt_dbxs[dbi].md_dcmp = cintcmp; + } else if (txn->mt_dbs[dbi].md_flags & MDB_REVERSEDUP) { + txn->mt_dbxs[dbi].md_dcmp = memnrcmp; + } else { + txn->mt_dbxs[dbi].md_dcmp = memncmp; + } + } else { + txn->mt_dbxs[dbi].md_dcmp = NULL; + } } int mdb_open(MDB_txn *txn, const char *name, unsigned int flags, MDB_dbi *dbi) @@ -3644,14 +4390,23 @@ int mdb_open(MDB_txn *txn, const char *name, unsigned int flags, MDB_dbi *dbi) int rc, dirty = 0; size_t len; + if (txn->mt_dbxs[FREE_DBI].md_cmp == NULL) { + mdb_default_cmp(txn, FREE_DBI); + } + /* main DB? */ if (!name) { *dbi = MAIN_DBI; if (flags & (MDB_DUPSORT|MDB_REVERSEKEY|MDB_INTEGERKEY)) txn->mt_dbs[MAIN_DBI].md_flags |= (flags & (MDB_DUPSORT|MDB_REVERSEKEY|MDB_INTEGERKEY)); + mdb_default_cmp(txn, MAIN_DBI); return MDB_SUCCESS; } + if (txn->mt_dbxs[MAIN_DBI].md_cmp == NULL) { + mdb_default_cmp(txn, MAIN_DBI); + } + /* Is the DB already open? */ len = strlen(name); for (i=2; imt_numdbs; i++) { @@ -3672,13 +4427,17 @@ int mdb_open(MDB_txn *txn, const char *name, unsigned int flags, MDB_dbi *dbi) /* Create if requested */ if (rc == MDB_NOTFOUND && (flags & MDB_CREATE)) { + MDB_cursor mc; 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; - rc = mdb_put0(txn, MAIN_DBI, &key, &data, F_SUBDATA); + mc.mc_txn = txn; + mc.mc_dbi = MAIN_DBI; + mc.mc_flags = 0; + rc = mdb_cursor_put(&mc, &key, &data, F_SUBDATA); dirty = 1; } @@ -3686,8 +4445,6 @@ int mdb_open(MDB_txn *txn, const char *name, unsigned int flags, MDB_dbi *dbi) if (rc == MDB_SUCCESS) { 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_cmp = NULL; - txn->mt_dbxs[txn->mt_numdbs].md_dcmp = NULL; 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; @@ -3695,6 +4452,7 @@ int mdb_open(MDB_txn *txn, const char *name, unsigned int flags, MDB_dbi *dbi) *dbi = txn->mt_numdbs; txn->mt_env->me_dbs[0][txn->mt_numdbs] = txn->mt_dbs[txn->mt_numdbs]; txn->mt_env->me_dbs[1][txn->mt_numdbs] = txn->mt_dbs[txn->mt_numdbs]; + mdb_default_cmp(txn, txn->mt_numdbs); txn->mt_numdbs++; } @@ -3746,3 +4504,5 @@ int mdb_set_relfunc(MDB_txn *txn, MDB_dbi dbi, MDB_rel_func *rel) txn->mt_dbxs[dbi].md_rel = rel; return MDB_SUCCESS; } + +/** @} */