#ifndef _WIN32
#include <pthread.h>
#ifdef MDB_USE_POSIX_SEM
+# define MDB_USE_HASH 1
#include <semaphore.h>
#endif
#endif
* @{
*/
#ifdef _WIN32
+#define MDB_USE_HASH 1
#define MDB_PIDLOCK 0
#define pthread_t DWORD
#define pthread_mutex_t HANDLE
(((mc)->mc_flags & C_SUB) ? -(int)(mc)->mc_dbi : (int)(mc)->mc_dbi)
/** @} */
- /** A default memory page size.
- * The actual size is platform-dependent, but we use this for
- * boot-strapping. We probably should not be using this any more.
- * The #GET_PAGESIZE() macro is used to get the actual size.
+ /** @brief The maximum size of a database page.
+ *
+ * This is 32k, since it must fit in #MDB_page.#mp_upper.
+ *
+ * LMDB will use database pages < OS pages if needed.
+ * That causes more I/O in write transactions: The OS must
+ * know (read) the whole page before writing a partial page.
*
* Note that we don't currently support Huge pages. On Linux,
* regular data files cannot use Huge pages, and in general
* pressure from other processes is high. So until OSs have
* actual paging support for Huge pages, they're not viable.
*/
-#define MDB_PAGESIZE 4096
+#define MAX_PAGESIZE 0x8000
/** The minimum number of keys required in a database page.
* Setting this to a larger value will place a smaller bound on the
*
* We require that keys all fit onto a regular page. This limit
* could be raised a bit further if needed; to something just
- * under #MDB_PAGESIZE / #MDB_MINKEYS.
+ * under (page size / #MDB_MINKEYS / 3).
*
* Note that data items in an #MDB_DUPSORT database are actually keys
* of a subDB, so they're also limited to this size.
txnid_t mm_txnid; /**< txnid that committed this page */
} MDB_meta;
- /** Buffer for a stack-allocated dirty page.
+ /** Buffer for a stack-allocated meta page.
* The members define size and alignment, and silence type
* aliasing warnings. They are not used directly; that could
* mean incorrectly using several union members in parallel.
*/
-typedef union MDB_pagebuf {
- char mb_raw[MDB_PAGESIZE];
+typedef union MDB_metabuf {
MDB_page mb_page;
struct {
char mm_pad[PAGEHDRSZ];
MDB_meta mm_meta;
} mb_metabuf;
-} MDB_pagebuf;
+} MDB_metabuf;
/** Auxiliary DB info.
* The information here is mostly static/read-only. There is
/** Have liveness lock in reader table */
#define MDB_LIVE_READER 0x08000000U
uint32_t me_flags; /**< @ref mdb_env */
- unsigned int me_psize; /**< size of a page, from #GET_PAGESIZE */
+ unsigned int me_psize; /**< DB page size, inited from me_os_psize */
+ unsigned int me_os_psize; /**< OS page size, from #GET_PAGESIZE */
unsigned int me_maxreaders; /**< size of the reader table */
unsigned int me_numreaders; /**< max numreaders set by this env */
MDB_dbi me_numdbs; /**< number of DBs opened */
char *me_map; /**< the memory map of the data file */
MDB_txninfo *me_txns; /**< the memory map of the lock file or NULL */
MDB_meta *me_metas[2]; /**< pointers to the two meta pages */
+ void *me_pbuf; /**< scratch area for DUPSORT put() */
MDB_txn *me_txn; /**< current write transaction */
size_t me_mapsize; /**< size of the data memory map */
off_t me_size; /**< current file size */
{
MDB_env *env = txn->mt_env;
MDB_page *ret = env->me_dpages;
- size_t sz = env->me_psize;
+ size_t psize = env->me_psize, sz = psize, off;
+ /* For #MDB_CLEANMEM, psize counts how much to init.
+ * For a single page alloc, we init everything after the page header.
+ * For multi-page, we init the final page; if the caller needed that
+ * many pages they will be filling in at least up to the last page.
+ */
if (num == 1) {
if (ret) {
VGMEMP_ALLOC(env, ret, sz);
env->me_dpages = ret->mp_next;
return ret;
}
+ psize -= off = PAGEHDRSZ;
} else {
sz *= num;
+ off = sz - psize;
}
if ((ret = malloc(sz)) != NULL) {
+ if (env->me_flags & MDB_CLEANMEM) {
+ memset((char *)ret + off, 0, psize);
+ ret->mp_pad = 0;
+ }
VGMEMP_ALLOC(env, ret, sz);
}
return ret;
int rc, maxfree_1pg = env->me_maxfree_1pg, more = 1;
txnid_t pglast = 0, head_id = 0;
pgno_t freecnt = 0, *free_pgs, *mop;
- ssize_t head_room = 0, total_room = 0, mop_len;
+ ssize_t head_room = 0, total_room = 0, mop_len, clean_limit;
mdb_cursor_init(&mc, txn, FREE_DBI, NULL);
return rc;
}
+ /* MDB_RESERVE cancels CLEANMEM in ovpage malloc (when no WRITEMAP) */
+ clean_limit = (env->me_flags & (MDB_CLEANMEM|MDB_WRITEMAP)) == MDB_CLEANMEM
+ ? maxfree_1pg : SSIZE_MAX;
+
for (;;) {
/* Come back here after each Put() in case freelist changed */
MDB_val key, data;
+ pgno_t *pgs;
+ ssize_t j;
/* If using records from freeDB which we have not yet
* deleted, delete them and any we reserved for me_pghead.
rc = mdb_cursor_put(&mc, &key, &data, MDB_RESERVE);
if (rc)
return rc;
- *(MDB_ID *)data.mv_data = 0; /* IDL is initially empty */
+ /* IDL is initially empty, zero out at least the length */
+ pgs = (pgno_t *)data.mv_data;
+ j = head_room > clean_limit ? head_room : 0;
+ do {
+ pgs[j] = 0;
+ } while (--j >= 0);
total_room += head_room;
}
static int
mdb_env_read_header(MDB_env *env, MDB_meta *meta)
{
- MDB_pagebuf pbuf;
+ MDB_metabuf pbuf;
MDB_page *p;
MDB_meta *m;
int i, rc, off;
+ enum { Size = sizeof(pbuf) };
/* We don't know the page size yet, so use a minimum value.
* Read both meta pages so we can use the latest one.
OVERLAPPED ov;
memset(&ov, 0, sizeof(ov));
ov.Offset = off;
- rc = ReadFile(env->me_fd,&pbuf,MDB_PAGESIZE,&len,&ov) ? (int)len : -1;
+ rc = ReadFile(env->me_fd, &pbuf, Size, &len, &ov) ? (int)len : -1;
if (rc == -1 && ErrCode() == ERROR_HANDLE_EOF)
rc = 0;
#else
- rc = pread(env->me_fd, &pbuf, MDB_PAGESIZE, off);
+ rc = pread(env->me_fd, &pbuf, Size, off);
#endif
- if (rc != MDB_PAGESIZE) {
+ if (rc != Size) {
if (rc == 0 && off == 0)
return ENOENT;
rc = rc < 0 ? (int) ErrCode() : MDB_INVALID;
mp->mm_last_pg = txn->mt_next_pgno - 1;
mp->mm_txnid = txn->mt_txnid;
if (!(env->me_flags & (MDB_NOMETASYNC|MDB_NOSYNC))) {
+ unsigned meta_size = env->me_psize;
rc = (env->me_flags & MDB_MAPASYNC) ? MS_ASYNC : MS_SYNC;
ptr = env->me_map;
- if (toggle)
- ptr += env->me_psize;
- if (MDB_MSYNC(ptr, env->me_psize, rc)) {
+ if (toggle) {
+#ifndef _WIN32 /* POSIX msync() requires ptr = start of OS page */
+ if (meta_size < env->me_os_psize)
+ meta_size += meta_size;
+ else
+#endif
+ ptr += meta_size;
+ }
+ if (MDB_MSYNC(ptr, meta_size, rc)) {
rc = ErrCode();
goto fail;
}
e->me_wmutex = SEM_FAILED;
#endif
e->me_pid = getpid();
+ GET_PAGESIZE(e->me_os_psize);
VGMEMP_CREATE(e,0,0);
*env = e;
return MDB_SUCCESS;
env->me_map = NULL;
return ErrCode();
}
- /* Turn off readahead. It's harmful when the DB is larger than RAM. */
+
+ if (flags & MDB_NORDAHEAD) {
+ /* Turn off readahead. It's harmful when the DB is larger than RAM. */
#ifdef MADV_RANDOM
- madvise(env->me_map, env->me_mapsize, MADV_RANDOM);
+ madvise(env->me_map, env->me_mapsize, MADV_RANDOM);
#else
#ifdef POSIX_MADV_RANDOM
- posix_madvise(env->me_map, env->me_mapsize, POSIX_MADV_RANDOM);
+ posix_madvise(env->me_map, env->me_mapsize, POSIX_MADV_RANDOM);
#endif /* POSIX_MADV_RANDOM */
#endif /* MADV_RANDOM */
+ }
#endif /* _WIN32 */
/* Can happen because the address argument to mmap() is just a
return i;
DPUTS("new mdbenv");
newenv = 1;
- GET_PAGESIZE(env->me_psize);
+ env->me_psize = env->me_os_psize;
+ if (env->me_psize > MAX_PAGESIZE)
+ env->me_psize = MAX_PAGESIZE;
} else {
env->me_psize = meta.mm_psize;
}
return rc;
}
-#if defined(_WIN32) || defined(MDB_USE_POSIX_SEM)
+#ifdef MDB_USE_HASH
/*
* hash_64 - 64 bit Fowler/Noll/Vo-0 FNV-1a hash code
*
* at runtime. Changing other flags requires closing the
* environment and re-opening it with the new flags.
*/
-#define CHANGEABLE (MDB_NOSYNC|MDB_NOMETASYNC|MDB_MAPASYNC)
-#define CHANGELESS (MDB_FIXEDMAP|MDB_NOSUBDIR|MDB_RDONLY|MDB_WRITEMAP|MDB_NOTLS|MDB_NOLOCK)
+#define CHANGEABLE (MDB_NOSYNC|MDB_NOMETASYNC|MDB_MAPASYNC|MDB_CLEANMEM)
+#define CHANGELESS (MDB_FIXEDMAP|MDB_NOSUBDIR|MDB_RDONLY|MDB_WRITEMAP| \
+ MDB_NOTLS|MDB_NOLOCK|MDB_NORDAHEAD)
int
mdb_env_open(MDB_env *env, const char *path, unsigned int flags, mdb_mode_t mode)
DPRINTF(("opened dbenv %p", (void *) env));
if (excl > 0) {
rc = mdb_env_share_locks(env, &excl);
+ if (rc)
+ goto leave;
}
+ if (!((flags & MDB_RDONLY) ||
+ (env->me_pbuf = calloc(1, env->me_psize))))
+ rc = ENOMEM;
}
leave:
for (i = env->me_maxdbs; --i > MAIN_DBI; )
free(env->me_dbxs[i].md_name.mv_data);
+ free(env->me_pbuf);
free(env->me_dbflags);
free(env->me_dbxs);
free(env->me_path);
rc = EINVAL;
} else {
MDB_page *mp = mc->mc_pg[mc->mc_top];
- if (!NUMKEYS(mp)) {
- mc->mc_ki[mc->mc_top] = 0;
+ int nkeys = NUMKEYS(mp);
+ if (!nkeys || mc->mc_ki[mc->mc_top] >= nkeys) {
+ mc->mc_ki[mc->mc_top] = nkeys;
rc = MDB_NOTFOUND;
break;
}
unsigned int flags)
{
enum { MDB_NO_ROOT = MDB_LAST_ERRCODE+10 }; /* internal code */
+ MDB_env *env = mc->mc_txn->mt_env;
MDB_node *leaf = NULL;
MDB_val xdata, *rdata, dkey;
- MDB_page *fp;
MDB_db dummy;
int do_sub = 0, insert = 0;
unsigned int mcount = 0, dcount = 0, nospill;
size_t nsize;
int rc, rc2;
- MDB_pagebuf pbuf;
char dbuf[MDB_MAXKEYSIZE+1];
unsigned int nflags;
DKBUF;
/* The key already exists */
if (rc == MDB_SUCCESS) {
+ MDB_page *fp, *mp;
+ MDB_val olddata;
+
/* there's only a key anyway, so this is a no-op */
if (IS_LEAF2(mc->mc_pg[mc->mc_top])) {
unsigned int ksize = mc->mc_db->md_pad;
return MDB_SUCCESS;
}
+more:
leaf = NODEPTR(mc->mc_pg[mc->mc_top], mc->mc_ki[mc->mc_top]);
+ olddata.mv_size = NODEDSZ(leaf);
+ olddata.mv_data = NODEDATA(leaf);
/* DB has dups? */
if (F_ISSET(mc->mc_db->md_flags, MDB_DUPSORT)) {
+ mp = fp = xdata.mv_data = env->me_pbuf;
+ mp->mp_pgno = mc->mc_pg[mc->mc_top]->mp_pgno;
+
/* Was a single item before, must convert now */
-more:
if (!F_ISSET(leaf->mn_flags, F_DUPDATA)) {
/* Just overwrite the current item */
if (flags == MDB_CURRENT)
goto current;
- dkey.mv_size = NODEDSZ(leaf);
- dkey.mv_data = NODEDATA(leaf);
+ dkey = olddata;
#if UINT_MAX < SIZE_MAX
if (mc->mc_dbx->md_dcmp == mdb_cmp_int && dkey.mv_size == sizeof(size_t))
#ifdef MISALIGNED_OK
/* create a fake page for the dup items */
memcpy(dbuf, dkey.mv_data, dkey.mv_size);
dkey.mv_data = dbuf;
- fp = (MDB_page *)&pbuf;
- fp->mp_pgno = mc->mc_pg[mc->mc_top]->mp_pgno;
fp->mp_flags = P_LEAF|P_DIRTY|P_SUBP;
fp->mp_lower = PAGEHDRSZ;
- fp->mp_upper = PAGEHDRSZ + dkey.mv_size + data->mv_size;
+ xdata.mv_size = PAGEHDRSZ + dkey.mv_size + data->mv_size;
if (mc->mc_db->md_flags & MDB_DUPFIXED) {
fp->mp_flags |= P_LEAF2;
fp->mp_pad = data->mv_size;
- fp->mp_upper += 2 * data->mv_size; /* leave space for 2 more */
+ xdata.mv_size += 2 * data->mv_size; /* leave space for 2 more */
} else {
- fp->mp_upper += 2 * sizeof(indx_t) + 2 * NODESIZE +
+ xdata.mv_size += 2 * (sizeof(indx_t) + NODESIZE) +
(dkey.mv_size & 1) + (data->mv_size & 1);
}
- mdb_node_del(mc->mc_pg[mc->mc_top], mc->mc_ki[mc->mc_top], 0);
- do_sub = 1;
- rdata = &xdata;
- xdata.mv_size = fp->mp_upper;
- xdata.mv_data = fp;
- flags |= F_DUPDATA;
- goto new_sub;
- }
- if (!F_ISSET(leaf->mn_flags, F_SUBDATA)) {
+ fp->mp_upper = xdata.mv_size;
+ } else if (leaf->mn_flags & F_SUBDATA) {
+ /* Data is on sub-DB, just store it */
+ flags |= F_DUPDATA|F_SUBDATA;
+ goto put_sub;
+ } else {
/* See if we need to convert from fake page to subDB */
- MDB_page *mp;
unsigned int offset;
unsigned int i;
uint16_t fp_flags;
- fp = NODEDATA(leaf);
- if (flags == MDB_CURRENT) {
-reuse:
+ fp = olddata.mv_data;
+ switch (flags) {
+ default:
+ if (!(mc->mc_db->md_flags & MDB_DUPFIXED)) {
+ offset = NODESIZE + sizeof(indx_t) + data->mv_size;
+ offset += offset & 1;
+ break;
+ }
+ offset = fp->mp_pad;
+ if (SIZELEFT(fp) < offset) {
+ offset *= 4; /* space for 4 more */
+ break;
+ }
+ /* FALLTHRU: Big enough MDB_DUPFIXED sub-page */
+ case MDB_CURRENT:
fp->mp_flags |= P_DIRTY;
- COPY_PGNO(fp->mp_pgno, mc->mc_pg[mc->mc_top]->mp_pgno);
+ COPY_PGNO(fp->mp_pgno, mp->mp_pgno);
mc->mc_xcursor->mx_cursor.mc_pg[0] = fp;
flags |= F_DUPDATA;
goto put_sub;
}
- if (mc->mc_db->md_flags & MDB_DUPFIXED) {
- offset = fp->mp_pad;
- if (SIZELEFT(fp) >= offset)
- goto reuse;
- offset *= 4; /* space for 4 more */
- } else {
- offset = NODESIZE + sizeof(indx_t) + data->mv_size;
- }
- offset += offset & 1;
fp_flags = fp->mp_flags;
- if (NODESIZE + sizeof(indx_t) + NODEKSZ(leaf) + NODEDSZ(leaf) +
- offset >= mc->mc_txn->mt_env->me_nodemax) {
+ xdata.mv_size = olddata.mv_size + offset;
+ if (NODESIZE + sizeof(indx_t) + NODEKSZ(leaf) + xdata.mv_size
+ >= env->me_nodemax) {
/* yes, convert it */
- dummy.md_flags = 0;
if (mc->mc_db->md_flags & MDB_DUPFIXED) {
dummy.md_pad = fp->mp_pad;
dummy.md_flags = MDB_DUPFIXED;
if (mc->mc_db->md_flags & MDB_INTEGERDUP)
dummy.md_flags |= MDB_INTEGERKEY;
+ } else {
+ dummy.md_pad = 0;
+ dummy.md_flags = 0;
}
dummy.md_depth = 1;
dummy.md_branch_pages = 0;
dummy.md_leaf_pages = 1;
dummy.md_overflow_pages = 0;
dummy.md_entries = NUMKEYS(fp);
- rdata = &xdata;
xdata.mv_size = sizeof(MDB_db);
xdata.mv_data = &dummy;
if ((rc = mdb_page_alloc(mc, 1, &mp)))
return rc;
- offset = mc->mc_txn->mt_env->me_psize - NODEDSZ(leaf);
+ offset = env->me_psize - olddata.mv_size;
flags |= F_DUPDATA|F_SUBDATA;
dummy.md_root = mp->mp_pgno;
fp_flags &= ~P_SUBP;
- } else {
- /* no, just grow it */
- rdata = &xdata;
- xdata.mv_size = NODEDSZ(leaf) + offset;
- xdata.mv_data = &pbuf;
- mp = (MDB_page *)&pbuf;
- mp->mp_pgno = mc->mc_pg[mc->mc_top]->mp_pgno;
- flags |= F_DUPDATA;
}
mp->mp_flags = fp_flags | P_DIRTY;
mp->mp_pad = fp->mp_pad;
if (IS_LEAF2(fp)) {
memcpy(METADATA(mp), METADATA(fp), NUMKEYS(fp) * fp->mp_pad);
} else {
- nsize = NODEDSZ(leaf) - fp->mp_upper;
- memcpy((char *)mp + mp->mp_upper, (char *)fp + fp->mp_upper, nsize);
+ memcpy((char *)mp + mp->mp_upper, (char *)fp + fp->mp_upper,
+ olddata.mv_size - fp->mp_upper);
for (i=0; i<NUMKEYS(fp); i++)
mp->mp_ptrs[i] = fp->mp_ptrs[i] + offset;
}
- mdb_node_del(mc->mc_pg[mc->mc_top], mc->mc_ki[mc->mc_top], 0);
- do_sub = 1;
- goto new_sub;
}
- /* data is on sub-DB, just store it */
- flags |= F_DUPDATA|F_SUBDATA;
- goto put_sub;
+
+ rdata = &xdata;
+ flags |= F_DUPDATA;
+ do_sub = 1;
+ mdb_node_del(mc->mc_pg[mc->mc_top], mc->mc_ki[mc->mc_top], 0);
+ goto new_sub;
}
current:
/* overflow page overwrites need special handling */
if (F_ISSET(leaf->mn_flags, F_BIGDATA)) {
MDB_page *omp;
pgno_t pg;
- unsigned psize = mc->mc_txn->mt_env->me_psize;
- int level, ovpages, dpages = OVPAGES(data->mv_size, psize);
+ int level, ovpages, dpages = OVPAGES(data->mv_size, env->me_psize);
- memcpy(&pg, NODEDATA(leaf), sizeof(pg));
+ memcpy(&pg, olddata.mv_data, sizeof(pg));
if ((rc2 = mdb_page_get(mc->mc_txn, pg, &omp, &level)) != 0)
return rc2;
ovpages = omp->mp_pages;
/* Is the ov page large enough? */
if (ovpages >= dpages) {
if (!(omp->mp_flags & P_DIRTY) &&
- (level || (mc->mc_txn->mt_env->me_flags & MDB_WRITEMAP)))
+ (level || (env->me_flags & MDB_WRITEMAP)))
{
rc = mdb_page_unspill(mc->mc_txn, omp, &omp);
if (rc)
*/
if (level > 1) {
/* It is writable only in a parent txn */
- size_t sz = (size_t) psize * ovpages, off;
+ size_t sz = (size_t) env->me_psize * ovpages, off;
MDB_page *np = mdb_page_malloc(mc->mc_txn, ovpages);
MDB_ID2 id2;
if (!np)
}
if ((rc2 = mdb_ovpage_free(mc, omp)) != MDB_SUCCESS)
return rc2;
- } else if (NODEDSZ(leaf) == data->mv_size) {
+ } else if (data->mv_size == olddata.mv_size) {
/* same size, just replace it. Note that we could
* also reuse this node if the new data is smaller,
* but instead we opt to shrink the node in that case.
*/
if (F_ISSET(flags, MDB_RESERVE))
- data->mv_data = NODEDATA(leaf);
+ data->mv_data = olddata.mv_data;
else if (data->mv_size)
- memcpy(NODEDATA(leaf), data->mv_data, data->mv_size);
+ memcpy(olddata.mv_data, data->mv_data, data->mv_size);
else
memcpy(NODEKEY(leaf), key->mv_data, key->mv_size);
goto done;
new_sub:
nflags = flags & NODE_ADD_FLAGS;
- nsize = IS_LEAF2(mc->mc_pg[mc->mc_top]) ? key->mv_size : mdb_leaf_size(mc->mc_txn->mt_env, key, rdata);
+ nsize = IS_LEAF2(mc->mc_pg[mc->mc_top]) ? key->mv_size : mdb_leaf_size(env, key, rdata);
if (SIZELEFT(mc->mc_pg[mc->mc_top]) < nsize) {
if (( flags & (F_DUPDATA|F_SUBDATA)) == F_DUPDATA )
nflags &= ~MDB_APPEND;
data[1].mv_size = mcount;
if (mcount < dcount) {
data[0].mv_data = (char *)data[0].mv_data + data[0].mv_size;
- leaf = NODEPTR(mc->mc_pg[mc->mc_top], mc->mc_ki[mc->mc_top]);
goto more;
}
}
mdb_cursor_del(MDB_cursor *mc, unsigned int flags)
{
MDB_node *leaf;
+ MDB_page *mp;
int rc;
if (mc->mc_txn->mt_flags & (MDB_TXN_RDONLY|MDB_TXN_ERROR))
if (!(mc->mc_flags & C_INITIALIZED))
return EINVAL;
+ if (mc->mc_ki[mc->mc_top] >= NUMKEYS(mc->mc_pg[mc->mc_top]))
+ return MDB_NOTFOUND;
+
if (!(flags & MDB_NOSPILL) && (rc = mdb_page_spill(mc, NULL, NULL)))
return rc;
if (rc)
return rc;
- leaf = NODEPTR(mc->mc_pg[mc->mc_top], mc->mc_ki[mc->mc_top]);
+ mp = mc->mc_pg[mc->mc_top];
+ leaf = NODEPTR(mp, mc->mc_ki[mc->mc_top]);
- if (!IS_LEAF2(mc->mc_pg[mc->mc_top]) && F_ISSET(leaf->mn_flags, F_DUPDATA)) {
+ if (!IS_LEAF2(mp) && F_ISSET(leaf->mn_flags, F_DUPDATA)) {
if (!(flags & MDB_NODUPDATA)) {
if (!F_ISSET(leaf->mn_flags, F_SUBDATA)) {
mc->mc_xcursor->mx_cursor.mc_pg[0] = NODEDATA(leaf);
} else {
MDB_cursor *m2;
/* shrink fake page */
- mdb_node_shrink(mc->mc_pg[mc->mc_top], mc->mc_ki[mc->mc_top]);
- leaf = NODEPTR(mc->mc_pg[mc->mc_top], mc->mc_ki[mc->mc_top]);
+ mdb_node_shrink(mp, mc->mc_ki[mc->mc_top]);
+ leaf = NODEPTR(mp, mc->mc_ki[mc->mc_top]);
mc->mc_xcursor->mx_cursor.mc_pg[0] = NODEDATA(leaf);
/* fix other sub-DB cursors pointed at this fake page */
for (m2 = mc->mc_txn->mt_cursors[mc->mc_dbi]; m2; m2=m2->mc_next) {
if (m2 == mc || m2->mc_snum < mc->mc_snum) continue;
- if (m2->mc_pg[mc->mc_top] == mc->mc_pg[mc->mc_top] &&
+ if (m2->mc_pg[mc->mc_top] == mp &&
m2->mc_ki[mc->mc_top] == mc->mc_ki[mc->mc_top])
m2->mc_xcursor->mx_cursor.mc_pg[0] = NODEDATA(leaf);
}
m3 = m2;
if (m3 == mc || m3->mc_snum < mc->mc_snum) continue;
if (m3->mc_pg[0] == mp) {
- m3->mc_pg[0] = mc->mc_pg[0];
- m3->mc_snum = 1;
- m3->mc_top = 0;
- m3->mc_ki[0] = m3->mc_ki[1];
+ int i;
+ m3->mc_snum--;
+ m3->mc_top--;
+ for (i=0; i<m3->mc_snum; i++) {
+ m3->mc_pg[i] = m3->mc_pg[i+1];
+ m3->mc_ki[i] = m3->mc_ki[i+1];
+ }
}
}
}
/* Adjust other cursors pointing to mp */
for (m2 = mc->mc_txn->mt_cursors[dbi]; m2; m2=m2->mc_next) {
- if (m2 == mc)
+ if (m2 == mc || m2->mc_snum < mc->mc_snum)
continue;
if (!(m2->mc_flags & C_INITIALIZED))
continue;
unsigned int nflags)
{
unsigned int flags;
- int rc = MDB_SUCCESS, ins_new = 0, new_root = 0, newpos = 1, did_split = 0;
+ int rc = MDB_SUCCESS, new_root = 0, did_split = 0;
indx_t newindx;
pgno_t pgno = 0;
- unsigned int i, j, split_indx, nkeys, pmax;
+ int i, j, split_indx, nkeys, pmax;
+ MDB_env *env = mc->mc_txn->mt_env;
MDB_node *node;
MDB_val sepkey, rkey, xdata, *rdata = &xdata;
- MDB_page *copy;
+ MDB_page *copy = NULL;
MDB_page *mp, *rp, *pp;
- unsigned int ptop;
+ int ptop;
MDB_cursor mn;
DKBUF;
mp = mc->mc_pg[mc->mc_top];
newindx = mc->mc_ki[mc->mc_top];
+ nkeys = NUMKEYS(mp);
- DPRINTF(("-----> splitting %s page %"Z"u and adding [%s] at index %i",
+ DPRINTF(("-----> splitting %s page %"Z"u and adding [%s] at index %i/%i",
IS_LEAF(mp) ? "leaf" : "branch", mp->mp_pgno,
- DKEY(newkey), mc->mc_ki[mc->mc_top]));
+ DKEY(newkey), mc->mc_ki[mc->mc_top], nkeys));
/* Create a right sibling. */
if ((rc = mdb_page_new(mc, mp->mp_flags, 1, &rp)))
sepkey = *newkey;
split_indx = newindx;
nkeys = 0;
- goto newsep;
- }
+ } else {
- nkeys = NUMKEYS(mp);
- split_indx = nkeys / 2;
- if (newindx < split_indx)
- newpos = 0;
-
- if (IS_LEAF2(rp)) {
- char *split, *ins;
- int x;
- unsigned int lsize, rsize, ksize;
- /* Move half of the keys to the right sibling */
- copy = NULL;
- x = mc->mc_ki[mc->mc_top] - split_indx;
- ksize = mc->mc_db->md_pad;
- split = LEAF2KEY(mp, split_indx, ksize);
- rsize = (nkeys - split_indx) * ksize;
- lsize = (nkeys - split_indx) * sizeof(indx_t);
- mp->mp_lower -= lsize;
- rp->mp_lower += lsize;
- mp->mp_upper += rsize - lsize;
- rp->mp_upper -= rsize - lsize;
- sepkey.mv_size = ksize;
- if (newindx == split_indx) {
- sepkey.mv_data = newkey->mv_data;
- } else {
- sepkey.mv_data = split;
- }
- if (x<0) {
- ins = LEAF2KEY(mp, mc->mc_ki[mc->mc_top], ksize);
- memcpy(rp->mp_ptrs, split, rsize);
- sepkey.mv_data = rp->mp_ptrs;
- memmove(ins+ksize, ins, (split_indx - mc->mc_ki[mc->mc_top]) * ksize);
- memcpy(ins, newkey->mv_data, ksize);
- mp->mp_lower += sizeof(indx_t);
- mp->mp_upper -= ksize - sizeof(indx_t);
+ split_indx = (nkeys+1) / 2;
+
+ if (IS_LEAF2(rp)) {
+ char *split, *ins;
+ int x;
+ unsigned int lsize, rsize, ksize;
+ /* Move half of the keys to the right sibling */
+ copy = NULL;
+ x = mc->mc_ki[mc->mc_top] - split_indx;
+ ksize = mc->mc_db->md_pad;
+ split = LEAF2KEY(mp, split_indx, ksize);
+ rsize = (nkeys - split_indx) * ksize;
+ lsize = (nkeys - split_indx) * sizeof(indx_t);
+ mp->mp_lower -= lsize;
+ rp->mp_lower += lsize;
+ mp->mp_upper += rsize - lsize;
+ rp->mp_upper -= rsize - lsize;
+ sepkey.mv_size = ksize;
+ if (newindx == split_indx) {
+ sepkey.mv_data = newkey->mv_data;
+ } else {
+ sepkey.mv_data = split;
+ }
+ if (x<0) {
+ ins = LEAF2KEY(mp, mc->mc_ki[mc->mc_top], ksize);
+ memcpy(rp->mp_ptrs, split, rsize);
+ sepkey.mv_data = rp->mp_ptrs;
+ memmove(ins+ksize, ins, (split_indx - mc->mc_ki[mc->mc_top]) * ksize);
+ memcpy(ins, newkey->mv_data, ksize);
+ mp->mp_lower += sizeof(indx_t);
+ mp->mp_upper -= ksize - sizeof(indx_t);
+ } else {
+ if (x)
+ memcpy(rp->mp_ptrs, split, x * ksize);
+ ins = LEAF2KEY(rp, x, ksize);
+ memcpy(ins, newkey->mv_data, ksize);
+ memcpy(ins+ksize, split + x * ksize, rsize - x * ksize);
+ rp->mp_lower += sizeof(indx_t);
+ rp->mp_upper -= ksize - sizeof(indx_t);
+ mc->mc_ki[mc->mc_top] = x;
+ mc->mc_pg[mc->mc_top] = rp;
+ }
} else {
- if (x)
- memcpy(rp->mp_ptrs, split, x * ksize);
- ins = LEAF2KEY(rp, x, ksize);
- memcpy(ins, newkey->mv_data, ksize);
- memcpy(ins+ksize, split + x * ksize, rsize - x * ksize);
- rp->mp_lower += sizeof(indx_t);
- rp->mp_upper -= ksize - sizeof(indx_t);
- mc->mc_ki[mc->mc_top] = x;
- mc->mc_pg[mc->mc_top] = rp;
- }
- goto newsep;
- }
+ int psize, nsize, k;
+ /* Maximum free space in an empty page */
+ pmax = env->me_psize - PAGEHDRSZ;
+ if (IS_LEAF(mp))
+ nsize = mdb_leaf_size(env, newkey, newdata);
+ else
+ nsize = mdb_branch_size(env, newkey);
+ nsize += nsize & 1;
- /* For leaf pages, check the split point based on what
- * fits where, since otherwise mdb_node_add can fail.
- *
- * This check is only needed when the data items are
- * relatively large, such that being off by one will
- * make the difference between success or failure.
- *
- * It's also relevant if a page happens to be laid out
- * such that one half of its nodes are all "small" and
- * the other half of its nodes are "large." If the new
- * item is also "large" and falls on the half with
- * "large" nodes, it also may not fit.
- */
- if (IS_LEAF(mp)) {
- unsigned int psize, nsize;
- /* Maximum free space in an empty page */
- pmax = mc->mc_txn->mt_env->me_psize - PAGEHDRSZ;
- nsize = mdb_leaf_size(mc->mc_txn->mt_env, newkey, newdata);
- if ((nkeys < 20) || (nsize > pmax/16)) {
- if (newindx <= split_indx) {
- psize = nsize;
- newpos = 0;
- for (i=0; i<split_indx; i++) {
- node = NODEPTR(mp, 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) {
- if (i <= newindx) {
- split_indx = newindx;
- if (i < newindx)
- newpos = 1;
+ /* grab a page to hold a temporary copy */
+ copy = mdb_page_malloc(mc->mc_txn, 1);
+ if (copy == NULL)
+ return ENOMEM;
+ copy->mp_pgno = mp->mp_pgno;
+ copy->mp_flags = mp->mp_flags;
+ copy->mp_lower = PAGEHDRSZ;
+ copy->mp_upper = env->me_psize;
+
+ /* prepare to insert */
+ for (i=0, j=0; i<nkeys; i++) {
+ if (i == newindx) {
+ copy->mp_ptrs[j++] = 0;
+ }
+ copy->mp_ptrs[j++] = mp->mp_ptrs[i];
+ }
+
+ /* When items are relatively large the split point needs
+ * to be checked, because being off-by-one will make the
+ * difference between success or failure in mdb_node_add.
+ *
+ * It's also relevant if a page happens to be laid out
+ * such that one half of its nodes are all "small" and
+ * the other half of its nodes are "large." If the new
+ * item is also "large" and falls on the half with
+ * "large" nodes, it also may not fit.
+ *
+ * As a final tweak, if the new item goes on the last
+ * spot on the page (and thus, onto the new page), bias
+ * the split so the new page is emptier than the old page.
+ * This yields better packing during sequential inserts.
+ */
+ if (nkeys < 20 || nsize > pmax/16 || newindx >= nkeys) {
+ /* Find split point */
+ psize = 0;
+ if (newindx <= split_indx || newindx >= nkeys) {
+ i = 0; j = 1;
+ k = newindx >= nkeys ? nkeys : split_indx+2;
+ } else {
+ i = nkeys; j = -1;
+ k = split_indx-1;
+ }
+ for (; i!=k; i+=j) {
+ if (i == newindx) {
+ psize += nsize;
+ node = NULL;
+ } else {
+ node = (MDB_node *)((char *)mp + copy->mp_ptrs[i]);
+ psize += NODESIZE + NODEKSZ(node) + sizeof(indx_t);
+ if (IS_LEAF(mp)) {
+ if (F_ISSET(node->mn_flags, F_BIGDATA))
+ psize += sizeof(pgno_t);
+ else
+ psize += NODEDSZ(node);
}
- else
- split_indx = i;
- break;
+ psize += psize & 1;
}
- }
- } else {
- psize = nsize;
- for (i=nkeys-1; i>=split_indx; i--) {
- node = NODEPTR(mp, 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) {
- if (i >= newindx) {
- split_indx = newindx;
- newpos = 0;
- } else
- split_indx = i+1;
+ if (psize > pmax || i == k-j) {
+ split_indx = i + (j<0);
break;
}
}
}
+ if (split_indx == newindx) {
+ sepkey.mv_size = newkey->mv_size;
+ sepkey.mv_data = newkey->mv_data;
+ } else {
+ node = (MDB_node *)((char *)mp + copy->mp_ptrs[split_indx]);
+ sepkey.mv_size = node->mn_ksize;
+ sepkey.mv_data = NODEKEY(node);
+ }
}
}
- /* First find the separating key between the split pages.
- * The case where newindx == split_indx is ambiguous; the
- * new item could go to the new page or stay on the original
- * page. If newpos == 1 it goes to the new page.
- */
- if (newindx == split_indx && newpos) {
- sepkey.mv_size = newkey->mv_size;
- sepkey.mv_data = newkey->mv_data;
- } else {
- node = NODEPTR(mp, split_indx);
- sepkey.mv_size = node->mn_ksize;
- sepkey.mv_data = NODEKEY(node);
- }
-
-newsep:
- DPRINTF(("separator is [%s]", DKEY(&sepkey)));
+ DPRINTF(("separator is %d [%s]", split_indx, DKEY(&sepkey)));
/* Copy separator key to the parent.
*/
- if (SIZELEFT(mn.mc_pg[ptop]) < mdb_branch_size(mc->mc_txn->mt_env, &sepkey)) {
+ if (SIZELEFT(mn.mc_pg[ptop]) < mdb_branch_size(env, &sepkey)) {
mn.mc_snum--;
mn.mc_top--;
did_split = 1;
return rc;
for (i=0; i<mc->mc_top; i++)
mc->mc_ki[i] = mn.mc_ki[i];
- goto done;
- }
- if (IS_LEAF2(rp)) {
- goto done;
- }
-
- /* Move half of the keys to the right sibling. */
+ } else if (!IS_LEAF2(mp)) {
+ /* Move nodes */
+ mc->mc_pg[mc->mc_top] = rp;
+ i = split_indx;
+ j = 0;
+ do {
+ if (i == newindx) {
+ rkey.mv_data = newkey->mv_data;
+ rkey.mv_size = newkey->mv_size;
+ if (IS_LEAF(mp)) {
+ rdata = newdata;
+ } else
+ pgno = newpgno;
+ flags = nflags;
+ /* Update index for the new key. */
+ mc->mc_ki[mc->mc_top] = j;
+ } else {
+ node = (MDB_node *)((char *)mp + copy->mp_ptrs[i]);
+ rkey.mv_data = NODEKEY(node);
+ rkey.mv_size = node->mn_ksize;
+ if (IS_LEAF(mp)) {
+ xdata.mv_data = NODEDATA(node);
+ xdata.mv_size = NODEDSZ(node);
+ rdata = &xdata;
+ } else
+ pgno = NODEPGNO(node);
+ flags = node->mn_flags;
+ }
- /* grab a page to hold a temporary copy */
- copy = mdb_page_malloc(mc->mc_txn, 1);
- if (copy == NULL)
- return ENOMEM;
+ if (!IS_LEAF(mp) && j == 0) {
+ /* First branch index doesn't need key data. */
+ rkey.mv_size = 0;
+ }
- copy->mp_pgno = mp->mp_pgno;
- copy->mp_flags = mp->mp_flags;
- copy->mp_lower = PAGEHDRSZ;
- copy->mp_upper = mc->mc_txn->mt_env->me_psize;
- mc->mc_pg[mc->mc_top] = copy;
- for (i = j = 0; i <= nkeys; j++) {
- if (i == split_indx) {
- /* Insert in right sibling. */
- /* Reset insert index for right sibling. */
- if (i != newindx || (newpos ^ ins_new)) {
+ rc = mdb_node_add(mc, j, &rkey, rdata, pgno, flags);
+ if (rc) {
+ /* return tmp page to freelist */
+ mdb_page_free(env, copy);
+ return rc;
+ }
+ if (i == nkeys) {
+ i = 0;
j = 0;
- mc->mc_pg[mc->mc_top] = rp;
+ mc->mc_pg[mc->mc_top] = copy;
+ } else {
+ i++;
+ j++;
+ }
+ } while (i != split_indx);
+
+ nkeys = NUMKEYS(copy);
+ for (i=0; i<nkeys; i++)
+ mp->mp_ptrs[i] = copy->mp_ptrs[i];
+ mp->mp_lower = copy->mp_lower;
+ mp->mp_upper = copy->mp_upper;
+ memcpy(NODEPTR(mp, nkeys-1), NODEPTR(copy, nkeys-1),
+ env->me_psize - copy->mp_upper);
+
+ /* reset back to original page */
+ if (newindx < split_indx) {
+ mc->mc_pg[mc->mc_top] = mp;
+ if (nflags & MDB_RESERVE) {
+ node = NODEPTR(mp, mc->mc_ki[mc->mc_top]);
+ if (!(node->mn_flags & F_BIGDATA))
+ newdata->mv_data = NODEDATA(node);
}
- }
-
- if (i == newindx && !ins_new) {
- /* Insert the original entry that caused the split. */
- rkey.mv_data = newkey->mv_data;
- rkey.mv_size = newkey->mv_size;
- if (IS_LEAF(mp)) {
- rdata = newdata;
- } else
- pgno = newpgno;
- flags = nflags;
-
- ins_new = 1;
-
- /* Update index for the new key. */
- mc->mc_ki[mc->mc_top] = j;
- } else if (i == nkeys) {
- break;
} else {
- node = NODEPTR(mp, i);
- rkey.mv_data = NODEKEY(node);
- rkey.mv_size = node->mn_ksize;
- if (IS_LEAF(mp)) {
- xdata.mv_data = NODEDATA(node);
- xdata.mv_size = NODEDSZ(node);
- rdata = &xdata;
- } else
- pgno = NODEPGNO(node);
- flags = node->mn_flags;
-
- i++;
- }
-
- if (!IS_LEAF(mp) && j == 0) {
- /* First branch index doesn't need key data. */
- rkey.mv_size = 0;
- }
-
- rc = mdb_node_add(mc, j, &rkey, rdata, pgno, flags);
- if (rc) break;
- }
-
- nkeys = NUMKEYS(copy);
- for (i=0; i<nkeys; i++)
- mp->mp_ptrs[i] = copy->mp_ptrs[i];
- mp->mp_lower = copy->mp_lower;
- mp->mp_upper = copy->mp_upper;
- memcpy(NODEPTR(mp, nkeys-1), NODEPTR(copy, nkeys-1),
- mc->mc_txn->mt_env->me_psize - copy->mp_upper);
-
- /* reset back to original page */
- if (newindx < split_indx || (!newpos && newindx == split_indx)) {
- mc->mc_pg[mc->mc_top] = mp;
- if (nflags & MDB_RESERVE) {
- node = NODEPTR(mp, mc->mc_ki[mc->mc_top]);
- if (!(node->mn_flags & F_BIGDATA))
- newdata->mv_data = NODEDATA(node);
- }
- } else {
- mc->mc_ki[ptop]++;
- /* Make sure mc_ki is still valid.
- */
- if (mn.mc_pg[ptop] != mc->mc_pg[ptop] &&
- mc->mc_ki[ptop] >= NUMKEYS(mc->mc_pg[ptop])) {
- for (i=0; i<ptop; i++) {
- mc->mc_pg[i] = mn.mc_pg[i];
- mc->mc_ki[i] = mn.mc_ki[i];
+ mc->mc_pg[mc->mc_top] = rp;
+ mc->mc_ki[ptop]++;
+ /* Make sure mc_ki is still valid.
+ */
+ if (mn.mc_pg[ptop] != mc->mc_pg[ptop] &&
+ mc->mc_ki[ptop] >= NUMKEYS(mc->mc_pg[ptop])) {
+ for (i=0; i<ptop; i++) {
+ mc->mc_pg[i] = mn.mc_pg[i];
+ mc->mc_ki[i] = mn.mc_ki[i];
+ }
+ mc->mc_pg[ptop] = mn.mc_pg[ptop];
+ mc->mc_ki[ptop] = mn.mc_ki[ptop] - 1;
}
- mc->mc_pg[ptop] = mn.mc_pg[ptop];
- mc->mc_ki[ptop] = mn.mc_ki[ptop] - 1;
}
+ /* return tmp page to freelist */
+ mdb_page_free(env, copy);
}
- /* return tmp page to freelist */
- mdb_page_free(mc->mc_txn->mt_env, copy);
-done:
{
/* Adjust other cursors pointing to mp */
MDB_cursor *m2, *m3;
}
}
}
+ DPRINTF(("mp left: %d, rp left: %d", SIZELEFT(mp), SIZELEFT(rp)));
return rc;
}
return MDB_SUCCESS;
}
+int
+mdb_env_get_fd(MDB_env *env, mdb_filehandle_t *arg)
+{
+ if (!env || !arg)
+ return EINVAL;
+
+ *arg = env->me_fd;
+ return MDB_SUCCESS;
+}
+
/** Common code for #mdb_stat() and #mdb_env_stat().
* @param[in] env the environment to operate in.
* @param[in] db the #MDB_db record containing the stats to return.