mdb.c

/** @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.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted only as authorized by the OpenLDAP
 * Public License.
 *
 * A copy of this license is available in the file LICENSE in the
 * top-level directory of the distribution or, alternatively, at
 * <http://www.OpenLDAP.org/license.html>.
 *
 * This code is derived from btree.c written by Martin Hedenfalk.
 *
 * Copyright (c) 2009, 2010 Martin Hedenfalk <martin@bzero.se>
 *
 * Permission to use, copy, modify, and distribute this software for any
 * purpose with or without fee is hereby granted, provided that the above
 * copyright notice and this permission notice appear in all copies.
 *
 * THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
 * WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
 * MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
 * ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
 * WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
 * ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
 * OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
 */
#include <sys/types.h>
#include <sys/stat.h>
#include <sys/param.h>
#ifdef _WIN32
#include <windows.h>
#else
#include <sys/uio.h>
#include <sys/mman.h>
#ifdef HAVE_SYS_FILE_H
#include <sys/file.h>
#endif
#include <fcntl.h>
#endif

#include <assert.h>
#include <errno.h>
#include <stddef.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unistd.h>

#ifndef _WIN32
#include <pthread.h>
#endif

#include "mdb.h"
#include "midl.h"

/** @defgroup internal	MDB Internals
 *	@{
 */
/** @defgroup compat	Windows Compatibility Macros
 *	A bunch of macros to minimize the amount of platform-specific ifdefs
 *	needed throughout the rest of the code. When the features this library
 *	needs are similar enough to POSIX to be hidden in a one-or-two line
 *	replacement, this macro approach is used.
 *	@{
 */
#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
	/** Lock the reader mutex.
	 */
#define LOCK_MUTEX_R(env)	pthread_mutex_lock(&env->me_txns->mti_mutex)
	/** Unlock the reader mutex.
	 */
#define UNLOCK_MUTEX_R(env)	pthread_mutex_unlock(&env->me_txns->mti_mutex)

	/** Lock the writer mutex.
	 *	Only a single write transaction is allowed at a time. Other writers
	 *	will block waiting for this mutex.
	 */
#define LOCK_MUTEX_W(env)	pthread_mutex_lock(&env->me_txns->mti_wmutex)
	/** Unlock the writer mutex.
	 */
#define UNLOCK_MUTEX_W(env)	pthread_mutex_unlock(&env->me_txns->mti_wmutex)

	/** Get the error code for the last failed system function.
	 */
#define	ErrCode()	errno

	/** An abstraction for a file handle.
	 *	On POSIX systems file handles are small integers. On Windows
	 *	they're opaque pointers.
	 */
#define	HANDLE	int

	/**	A value for an invalid file handle.
	 *	Mainly used to initialize file variables and signify that they are
	 *	unused.
	 */
#define INVALID_HANDLE_VALUE	-1

	/** Get the size of a memory page for the system.
	 *	This is the basic size that the platform's memory manager uses, and is
	 *	fundamental to the use of memory-mapped files.
	 */
#define	GetPageSize(x)	(x) = sysconf(_SC_PAGE_SIZE)
#endif

/** @} */

#ifndef _WIN32
/**	A flag for opening a file and requesting synchronous data writes.
 *	This is only used when writing a meta page. It's not strictly needed;
 *	we could just do a normal write and then immediately perform a flush.
 *	But if this flag is available it saves us an extra system call.
 *
 *	@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.
 */
#ifndef MDB_DSYNC
# define MDB_DSYNC	O_DSYNC
#endif
#endif

	/** A page number in the database.
	 *	Note that 64 bit page numbers are overkill, since pages themselves
	 *	already represent 12-13 bits of addressable memory, and the OS will
	 *	always limit applications to a maximum of 63 bits of address space.
	 *
	 *	@note In the #MDB_node structure, we only store 48 bits of this value,
	 *	which thus limits us to only 60 bits of addressable data.
	 */
typedef ULONG		pgno_t;

/** @defgroup debug	Debug Macros
 *	@{
 */
#ifndef DEBUG
	/**	Enable debug output.
	 *	Set this to 1 for copious tracing. Set to 2 to add dumps of all IDLs
	 *	read from and written to the database (used for free space management).
	 */
#define DEBUG 0
#endif

#if !(__STDC_VERSION__ >= 199901L || defined(__GNUC__))
# define DPRINTF	(void)	/* Vararg macros may be unsupported */
#elif DEBUG
	/**	Print a debug message with printf formatting. */
# define DPRINTF(fmt, ...)	/**< Requires 2 or more args */ \
	fprintf(stderr, "%s:%d:(%p) " fmt "\n", __func__, __LINE__, pthread_self(), __VA_ARGS__)
#else
# define DPRINTF(fmt, ...)	((void) 0)
#endif
	/**	Print a debug string.
	 *	The string is printed literally, with no format processing.
	 */
#define DPUTS(arg)	DPRINTF("%s", arg)
/** @} */

	/** 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 #GetPageSize() macro is used to get the actual size.
	 *
	 *	Note that we don't currently support Huge pages. On Linux,
	 *	regular data files cannot use Huge pages, and in general
	 *	Huge pages aren't actually pageable. We rely on the OS
	 *	demand-pager to read our data and page it out when memory
	 *	pressure from other processes is high. So until OSs have
	 *	actual paging support for Huge pages, they're not viable.
	 */
#define PAGESIZE	 4096

	/** The minimum number of keys required in a database page.
	 *	Setting this to a larger value will place a smaller bound on the
	 *	maximum size of a data item. Data items larger than this size will
	 *	be pushed into overflow pages instead of being stored directly in
	 *	the B-tree node. This value used to default to 4. With a page size
	 *	of 4096 bytes that meant that any item larger than 1024 bytes would
	 *	go into an overflow page. That also meant that on average 2-3KB of
	 *	each overflow page was wasted space. The value cannot be lower than
	 *	2 because then there would no longer be a tree structure. With this
	 *	value, items larger than 2KB will go into overflow pages, and on
	 *	average only 1KB will be wasted.
	 */
#define MDB_MINKEYS	 2

	/**	A stamp that identifies a file as an MDB file.
	 *	There's nothing special about this value other than that it is easily
	 *	recognizable, and it will reflect any byte order mismatches.
	 */
#define MDB_MAGIC	 0xBEEFC0DE

	/**	The version number for a database's file format. */
#define MDB_VERSION	 1

	/**	The maximum size of a key in the database.
	 *	While data items have essentially unbounded size, we require that
	 *	keys all fit onto a regular page. This limit could be raised a bit
	 *	further if needed; to something just under #PAGESIZE / #MDB_MINKEYS.
	 */
#define MAXKEYSIZE	 511

#if DEBUG
	/**	A key buffer.
	 *	@ingroup debug
	 *	This is used for printing a hex dump of a key's contents.
	 */
#define DKBUF	char kbuf[(MAXKEYSIZE*2+1)]
	/**	Display a key in hex.
	 *	@ingroup debug
	 *	Invoke a function to display a key in hex.
	 */
#define	DKEY(x)	mdb_dkey(x, kbuf)
#else
#define	DKBUF
#define DKEY(x)
#endif

/**	@defgroup lazylock	Lazy Locking
 *	Macros for locks that are't actually needed.
 *	The DB view is always consistent because all writes are wrapped in
 *	the wmutex. Finer-grained locks aren't necessary.
 *	@{
 */
#ifndef	LAZY_LOCKS
	/**	Use lazy locking. I.e., don't lock these accesses at all. */
#define	LAZY_LOCKS	1
#endif
#if	LAZY_LOCKS
	/** Grab the reader lock */
#define	LAZY_MUTEX_LOCK(x)
	/** Release the reader lock */
#define	LAZY_MUTEX_UNLOCK(x)
	/** Release the DB table reader/writer lock */
#define	LAZY_RWLOCK_UNLOCK(x)
	/** Grab the DB table write lock */
#define	LAZY_RWLOCK_WRLOCK(x)
	/** Grab the DB table read lock */
#define	LAZY_RWLOCK_RDLOCK(x)
	/** Declare the DB table rwlock */
#define	LAZY_RWLOCK_DEF(x)
	/** Initialize the DB table rwlock */
#define	LAZY_RWLOCK_INIT(x,y)
	/**	Destroy the DB table rwlock */
#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
/** @} */

	/** An invalid page number.
	 *	Mainly used to denote an empty tree.
	 */
#define P_INVALID	 (~0UL)

	/** Test if a flag \b f is set in a flag word \b w. */
#define F_ISSET(w, f)	 (((w) & (f)) == (f))

	/**	Used for offsets within a single page.
	 *	Since memory pages are typically 4 or 8KB in size, 12-13 bits,
	 *	this is plenty.
	 */
typedef uint16_t	 indx_t;

	/**	Default size of memory map.
	 *	This is certainly too small for any actual applications. Apps should always set
	 *	the size explicitly using #mdb_env_set_mapsize().
	 */
#define DEFAULT_MAPSIZE	1048576

/**	@defgroup readers	Reader Lock Table
 *	Readers don't acquire any locks for their data access. Instead, they
 *	simply record their transaction ID in the reader table. The reader
 *	mutex is needed just to find an empty slot in the reader table. The
 *	slot's address is saved in thread-specific data so that subsequent read
 *	transactions started by the same thread need no further locking to proceed.
 *
 *	Since the database uses multi-version concurrency control, readers don't
 *	actually need any locking. This table is used to keep track of which
 *	readers are using data from which old transactions, so that we'll know
 *	when a particular old transaction is no longer in use, Old transactions
 *	that have freed any data pages can then have their freed pages reclaimed
 *	for use by a later write transaction.
 *
 *	The lock table is constructed such that reader slots are aligned with the
 *	processor's cache line size. Any slot is only ever used by one thread.
 *	This alignment guarantees that there will be no contention or cache
 *	thrashing as threads update their own slot info, and also eliminates
 *	any need for locking when accessing a slot.
 *
 *	A writer thread will scan every slot in the table to determine the oldest
 *	outstanding reader transaction. Any freed pages older than this will be
 *	reclaimed by the writer. The writer doesn't use any locks when scanning
 *	this table. This means that there's no guarantee that the writer will
 *	see the most up-to-date reader info, but that's not required for correct
 *	operation - all we need is to know the upper bound on the oldest reader,
 *	we don't care at all about the newest reader. So the only consequence of
 *	reading stale information here is that old pages might hang around a
 *	while longer before being reclaimed. That's actually good anyway, because
 *	the longer we delay reclaiming old pages, the more likely it is that a
 *	string of contiguous pages can be found after coalescing old pages from
 *	many old transactions together.
 *
 *	@todo We don't actually do such coalescing yet, we grab pages from one
 *	old transaction at a time.
 *	@{
 */
	/**	Number of slots in the reader table.
	 *	This value was chosen somewhat arbitrarily. 126 readers plus a
	 *	couple mutexes fit exactly into 8KB on my development machine.
	 *	Applications should set the table size using #mdb_env_set_maxreaders().
	 */
#define DEFAULT_READERS	126

	/**	The size of a CPU cache line in bytes. We want our lock structures
	 *	aligned to this size to avoid false cache line sharing in the
	 *	lock table.
	 *	This value works for most CPUs. For Itanium this should be 128.
	 */
#ifndef CACHELINE
#define CACHELINE	64
#endif

	/**	The information we store in a single slot of the reader table.
	 *	In addition to a transaction ID, we also record the process and
	 *	thread ID that owns a slot, so that we can detect stale information,
	 *	e.g. threads or processes that went away without cleaning up.
	 *	@note We currently don't check for stale records. We simply re-init
	 *	the table when we know that we're the only process opening the
	 *	lock file.
	 */
typedef struct MDB_rxbody {
	/**	The current Transaction ID when this transaction began.
	 *	Multiple readers that start at the same time will probably have the
	 *	same ID here. Again, it's not important to exclude them from
	 *	anything; all we need to know is which version of the DB they
	 *	started from so we can avoid overwriting any data used in that
	 *	particular version.
	 */
	ULONG		mrb_txnid;
	/** The process ID of the process owning this reader txn. */
	pid_t		mrb_pid;
	/** The thread ID of the thread owning this txn. */
	pthread_t	mrb_tid;
} MDB_rxbody;

	/** The actual reader record, with cacheline padding. */
typedef struct MDB_reader {
	union {
		MDB_rxbody mrx;
		/** shorthand for mrb_txnid */
#define	mr_txnid	mru.mrx.mrb_txnid
#define	mr_pid	mru.mrx.mrb_pid
#define	mr_tid	mru.mrx.mrb_tid
		/** cache line alignment */
		char pad[(sizeof(MDB_rxbody)+CACHELINE-1) & ~(CACHELINE-1)];
	} mru;
} MDB_reader;

	/** The header for the reader table.
	 *	The table resides in a memory-mapped file. (This is a different file
	 *	than is used for the main database.)
	 *
	 *	For POSIX the actual mutexes reside in the shared memory of this
	 *	mapped file. On Windows, mutexes are named objects allocated by the
	 *	kernel; we store the mutex names in this mapped file so that other
	 *	processes can grab them. This same approach will also be used on
	 *	MacOSX/Darwin (using named semaphores) since MacOSX doesn't support
	 *	process-shared POSIX mutexes.
	 */
typedef struct MDB_txbody {
		/** Stamp identifying this as an MDB lock file. It must be set
		 *	to #MDB_MAGIC. */
	uint32_t	mtb_magic;
		/** Version number of this lock file. Must be set to #MDB_VERSION. */
	uint32_t	mtb_version;
#ifdef _WIN32
	char	mtb_rmname[32];
#else
		/** Mutex protecting access to this table.
		 *	This is the reader lock that #LOCK_MUTEX_R acquires.
		 */
	pthread_mutex_t	mtb_mutex;
#endif
		/**	The ID of the last transaction committed to the database.
		 *	This is recorded here only for convenience; the value can always
		 *	be determined by reading the main database meta pages.
		 */
	ULONG		mtb_txnid;
		/** The number of slots that have been used in the reader table.
		 *	This always records the maximum count, it is not decremented
		 *	when readers release their slots.
		 */
	uint32_t	mtb_numreaders;
		/**	The ID of the most recent meta page in the database.
		 *	This is recorded here only for convenience; the value can always
		 *	be determined by reading the main database meta pages.
		 */
	uint32_t	mtb_me_toggle;
} MDB_txbody;

	/** The actual reader table definition. */
typedef struct MDB_txninfo {
	union {
		MDB_txbody mtb;
#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];
} MDB_txninfo;
/** @} */

/** Common header for all page types.
 * Overflow pages occupy a number of contiguous pages with no
 * headers on any page after the first.
 */
typedef struct MDB_page {
#define	mp_pgno	mp_p.p_pgno
#define	mp_next	mp_p.p_next
	union padded {
		pgno_t		p_pgno;	/**< page number */
		void *		p_next;	/**< for in-memory list of freed structs */
	} mp_p;
#define	P_BRANCH	 0x01		/**< branch page */
#define	P_LEAF		 0x02		/**< leaf page */
#define	P_OVERFLOW	 0x04		/**< overflow page */
#define	P_META		 0x08		/**< meta page */
#define	P_DIRTY		 0x10		/**< dirty page */
#define	P_LEAF2		 0x20		/**< for #MDB_DUPFIXED records */
	uint32_t	mp_flags;
#define mp_lower	mp_pb.pb.pb_lower
#define mp_upper	mp_pb.pb.pb_upper
#define mp_pages	mp_pb.pb_pages
	union page_bounds {
		struct {
			indx_t		pb_lower;		/**< lower bound of free space */
			indx_t		pb_upper;		/**< upper bound of free space */
		} pb;
		uint32_t	pb_pages;	/**< number of overflow pages */
	} mp_pb;
	indx_t		mp_ptrs[1];		/**< dynamic size */
} MDB_page;

	/** Size of the page header, excluding dynamic data at the end */
#define PAGEHDRSZ	 ((unsigned) offsetof(MDB_page, mp_ptrs))

	/** Address of first usable data byte in a page, after the header */
#define METADATA(p)	 ((void *)((char *)(p) + PAGEHDRSZ))

	/** Number of nodes on a page */
#define NUMKEYS(p)	 (((p)->mp_lower - PAGEHDRSZ) >> 1)

	/** The amount of space remaining in the page */
#define SIZELEFT(p)	 (indx_t)((p)->mp_upper - (p)->mp_lower)

	/** The percentage of space used in the page, in tenths of a percent. */
#define PAGEFILL(env, p) (1000L * ((env)->me_psize - PAGEHDRSZ - SIZELEFT(p)) / \
				((env)->me_psize - PAGEHDRSZ))
	/** The minimum page fill factor, in tenths of a percent.
	 *	Pages emptier than this are candidates for merging.
	 */
#define FILL_THRESHOLD	 250

	/** Test if a page is a leaf page */
#define IS_LEAF(p)	 F_ISSET((p)->mp_flags, P_LEAF)
	/** Test if a page is a LEAF2 page */
#define IS_LEAF2(p)	 F_ISSET((p)->mp_flags, P_LEAF2)
	/** Test if a page is a branch page */
#define IS_BRANCH(p)	 F_ISSET((p)->mp_flags, P_BRANCH)
	/** Test if a page is an overflow page */
#define IS_OVERFLOW(p)	 F_ISSET((p)->mp_flags, P_OVERFLOW)

	/** The number of overflow pages needed to store the given size. */
#define OVPAGES(size, psize)	((PAGEHDRSZ-1 + (size)) / (psize) + 1)

	/** Header for a single key/data pair within a page.
	 * We guarantee 2-byte alignment for nodes.
	 */
typedef struct MDB_node {
	/** 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;			/**< part of dsize or pgno */
	unsigned short	mn_flags;		/**< flags for special node types */
#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];			/**< key and data are appended here */
} MDB_node;

	/** Size of the node header, excluding dynamic data at the end */
#define NODESIZE	 offsetof(MDB_node, mn_data)

	/** Size of a node in a branch page with a given key.
	 *	This is just the node header plus the key, there is no data.
	 */
#define INDXSIZE(k)	 (NODESIZE + ((k) == NULL ? 0 : (k)->mv_size))

	/** Size of a node in a leaf page with a given key and data.
	 *	This is node header plus key plus data size.
	 */
#define LEAFSIZE(k, d)	 (NODESIZE + (k)->mv_size + (d)->mv_size)

	/** Address of node \b i in page \b p */
#define NODEPTR(p, i)	 ((MDB_node *)((char *)(p) + (p)->mp_ptrs[i]))

	/** Address of the key for the node */
#define NODEKEY(node)	 (void *)((node)->mn_data)

	/** Address of the data for a node */
#define NODEDATA(node)	 (void *)((char *)(node)->mn_data + (node)->mn_ksize)

	/** Get the page number pointed to by a branch node */
#if LONG_MAX == 0x7fffffff
#define NODEPGNO(node)	 ((node)->mn_lo | ((node)->mn_hi << 16))
	/** Set the page number in a branch node */
#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))
	/** Set the page number in a branch node */
#define SETPGNO(node,pgno)	do { \
	(node)->mn_lo = (pgno) & 0xffff; (node)->mn_hi = (pgno) >> 16; \
	(node)->mn_flags = (pgno) >> 32; } while(0)
#endif

	/** Get the size of the data in a leaf node */
#define NODEDSZ(node)	 ((node)->mn_lo | ((unsigned)(node)->mn_hi << 16))
	/** Set the size of the data for a leaf node */
#define SETDSZ(node,size)	do { \
	(node)->mn_lo = (size) & 0xffff; (node)->mn_hi = (size) >> 16;} while(0)
	/** The size of a key in a node */
#define NODEKSZ(node)	 ((node)->mn_ksize)

	/** The address of a key in a LEAF2 page.
	 *	LEAF2 pages are used for #MDB_DUPFIXED sorted-duplicate sub-DBs.
	 *	There are no node headers, keys are stored contiguously.
	 */
#define LEAF2KEY(p, i, ks)	((char *)(p) + PAGEHDRSZ + ((i)*(ks)))

	/** Set the \b node's key into \b key, if requested. */
#define MDB_SET_KEY(node, key)	if (key!=NULL) {(key)->mv_size = NODEKSZ(node); (key)->mv_data = NODEKEY(node);}

	/** Information about a single database in the environment. */
typedef struct MDB_db {
	uint32_t	md_pad;		/**< also ksize for LEAF2 pages */
	uint16_t	md_flags;	/**< @ref mdb_open */
	uint16_t	md_depth;	/**< depth of this tree */
	ULONG		md_branch_pages;	/**< number of internal pages */
	ULONG		md_leaf_pages;		/**< number of leaf pages */
	ULONG		md_overflow_pages;	/**< number of overflow pages */
	ULONG		md_entries;		/**< number of data items */
	pgno_t		md_root;		/**< the root page of this tree */
} MDB_db;

	/** Handle for the DB used to track free pages. */
#define	FREE_DBI	0
	/** Handle for the default DB. */
#define	MAIN_DBI	1

	/** Meta page content. */
typedef struct MDB_meta {
		/** Stamp identifying this as an MDB data file. It must be set
		 *	to #MDB_MAGIC. */
	uint32_t	mm_magic;
		/** Version number of this lock file. Must be set to #MDB_VERSION. */
	uint32_t	mm_version;
	void		*mm_address;		/**< address for fixed mapping */
	size_t		mm_mapsize;			/**< size of mmap region */
	MDB_db		mm_dbs[2];			/**< first is free space, 2nd is main db */
	/** The size of pages used in this DB */
#define	mm_psize	mm_dbs[0].md_pad
	/** Any persistent environment flags. @ref mdb_env */
#define	mm_flags	mm_dbs[0].md_flags
	pgno_t		mm_last_pg;			/**< last used page in file */
	ULONG		mm_txnid;			/**< txnid that committed this page */
} MDB_meta;

	/** Auxiliary DB info.
	 *	The information here is mostly static/read-only. There is
	 *	only a single copy of this record in the environment.
	 *	The \b md_dirty flag is not read-only, but only a write
	 *	transaction can ever update it, and only write transactions
	 *	need to worry about it.
	 */
typedef struct MDB_dbx {
	MDB_val		md_name;		/**< name of the database */
	MDB_cmp_func	*md_cmp;	/**< function for comparing keys */
	MDB_cmp_func	*md_dcmp;	/**< function for comparing data items */
	MDB_rel_func	*md_rel;	/**< user relocate function */
	MDB_dbi	md_parent;			/**< parent DB of a sub-DB */
	unsigned int	md_dirty;	/**< TRUE if DB was written in this txn */
} MDB_dbx;

	/** A database transaction.
	 *	Every operation requires a transaction handle.
	 */
struct MDB_txn {
	pgno_t		mt_next_pgno;	/**< next unallocated page */
	/** The ID of this transaction. IDs are integers incrementing from 1.
	 *	Only committed write transactions increment the ID. If a transaction
	 *	aborts, the ID may be re-used by the next writer.
	 */
	ULONG		mt_txnid;
	MDB_env		*mt_env;		/**< the DB environment */
	/** The list of pages that became unused during this transaction.
	 *	This is an #IDL.
	 */
	pgno_t		*mt_free_pgs;
	union {
		ID2L	dirty_list;	/**< modified pages */
		MDB_reader	*reader;	/**< this thread's slot in the reader table */
	} mt_u;
	/** Array of records for each DB known in the environment. */
	MDB_dbx		*mt_dbxs;
	/** Array of MDB_db records for each known DB */
	MDB_db		*mt_dbs;
	/**	Number of DB records in use. This number only ever increments;
	 *	we don't decrement it when individual DB handles are closed.
	 */
	unsigned int	mt_numdbs;

#define MDB_TXN_RDONLY		0x01		/**< read-only transaction */
#define MDB_TXN_ERROR		0x02		/**< an error has occurred */
	unsigned int	mt_flags;
	/** Tracks which of the two meta pages was used at the start
	 * 	of this transaction.
	 */
	unsigned int	mt_toggle;
};

/** Enough space for 2^32 nodes with minimum of 2 keys per node. I.e., plenty.
 * At 4 keys per node, enough for 2^64 nodes, so there's probably no need to
 * raise this on a 64 bit machine.
 */
#define CURSOR_STACK		 32

struct MDB_xcursor;

	/** Cursors are used for all DB operations */
struct MDB_cursor {
	/** Context used for databases with #MDB_DUPSORT, otherwise NULL */
	struct MDB_xcursor	*mc_xcursor;
	/** The transaction that owns this cursor */
	MDB_txn		*mc_txn;
	/** The database handle this cursor operates on */
	MDB_dbi		mc_dbi;
	unsigned short 	mc_snum;	/**< number of pushed pages */
	unsigned short	mc_top;		/**< index of top page, mc_snum-1 */
	unsigned int	mc_flags;
#define C_INITIALIZED	0x01	/**< cursor has been initialized and is valid */
#define C_EOF	0x02			/**< No more data */
#define C_XDIRTY	0x04		/**< @deprecated mc_xcursor needs to be flushed */
	MDB_page	*mc_pg[CURSOR_STACK];	/**< stack of pushed pages */
	indx_t		mc_ki[CURSOR_STACK];	/**< stack of page indices */
};

	/** Context for sorted-dup records.
	 *	We could have gone to a fully recursive design, with arbitrarily
	 *	deep nesting of sub-databases. But for now we only handle these
	 *	levels - main DB, optional sub-DB, sorted-duplicate DB.
	 */
typedef struct MDB_xcursor {
	/** A sub-cursor for traversing the Dup DB */
	MDB_cursor mx_cursor;
	/** A fake transaction struct for pointing to our own table
	 *	of DB info.
	 */
	MDB_txn mx_txn;
	/**	Our private DB information tables. Slots 0 and 1 are always
	 *	copies of the corresponding slots in the main transaction. These
	 *	hold the FREEDB and MAINDB, respectively. If the main cursor is
	 *	on a sub-database, that will be copied to slot 2, and the duplicate
	 *	database info will be in slot 3. If the main cursor is on the MAINDB
	 *	then the duplicate DB info will be in slot 2 and slot 3 will be unused.
	 */
	MDB_dbx	mx_dbxs[4];
	/** MDB_db table */
	MDB_db	mx_dbs[4];
} MDB_xcursor;

	/** A set of pages freed by an earlier transaction. */
typedef struct MDB_oldpages {
	/** Usually we only read one record from the FREEDB at a time, but
	 *	in case we read more, this will chain them together.
	 */
	struct MDB_oldpages *mo_next;
	/**	The ID of the transaction in which these pages were freed. */
	ULONG		mo_txnid;
	/** An #IDL of the pages */
	pgno_t		mo_pages[1];	/* dynamic */
} MDB_oldpages;

	/** The database environment. */
struct MDB_env {
	HANDLE		me_fd;		/**< The main data file */
	HANDLE		me_lfd;		/**< The lock file */
	HANDLE		me_mfd;			/**< just for writing the meta pages */
	/** Failed to update the meta page. Probably an I/O error. */
#define	MDB_FATAL_ERROR	0x80000000U
	uint32_t 	me_flags;
	uint32_t	me_extrapad;	/**< unused for now */
	unsigned int	me_maxreaders;	/**< size of the reader table */
	unsigned int	me_numdbs;		/**< number of DBs opened */
	unsigned int	me_maxdbs;		/**< size of the DB table */
	char		*me_path;		/**< path to the DB files */
	char		*me_map;		/**< the memory map of the data file */
	MDB_txninfo	*me_txns;		/**< the memory map of the lock file */
	MDB_meta	*me_metas[2];	/**< pointers to the two meta pages */
	MDB_txn		*me_txn;		/**< current write transaction */
	size_t		me_mapsize;		/**< size of the data memory map */
	off_t		me_size;		/**< current file size */
	pgno_t		me_maxpg;		/**< me_mapsize / me_psize */
	unsigned int	me_psize;	/**< size of a page, from #GetPageSize */
	unsigned int	me_db_toggle;	/**< which DB table is current */
	MDB_dbx		*me_dbxs;		/**< array of static DB info */
	MDB_db		*me_dbs[2];		/**< two arrays of MDB_db info */
	MDB_oldpages *me_pghead;	/**< list of old page records */
	pthread_key_t	me_txkey;	/**< thread-key for readers */
	MDB_page	*me_dpages;		/**< list of malloc'd blocks for re-use */
	/** IDL of pages that became unused in a write txn */
	pgno_t		me_free_pgs[MDB_IDL_UM_SIZE];
	/** ID2L of pages that were written during a write txn */
	ID2			me_dirty_list[MDB_IDL_UM_SIZE];
	/** rwlock for the DB tables, if #LAZY_LOCKS is false */
	LAZY_RWLOCK_DEF(me_dblock);
#ifdef _WIN32
	HANDLE		me_rmutex;		/* Windows mutexes don't reside in shared mem */
	HANDLE		me_wmutex;
#endif
};
	/** max number of pages to commit in one writev() call */
#define MDB_COMMIT_PAGES	 64

static MDB_page *mdb_alloc_page(MDB_cursor *mc, int num);
static int 		mdb_touch(MDB_cursor *mc);

static int  mdb_search_page_root(MDB_cursor *mc,
			    MDB_val *key, int modify);
static int  mdb_search_page(MDB_cursor *mc,
			    MDB_val *key, int modify);

static int  mdb_env_read_header(MDB_env *env, MDB_meta *meta);
static int  mdb_env_read_meta(MDB_env *env, int *which);
static int  mdb_env_write_meta(MDB_txn *txn);
static int  mdb_get_page(MDB_txn *txn, pgno_t pgno, MDB_page **mp);

static MDB_node *mdb_search_node(MDB_cursor *mc, MDB_val *key, int *exactp);
static int  mdb_add_node(MDB_cursor *mc, 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_cursor *mc, MDB_node *leaf);
static int  mdb_read_data(MDB_txn *txn, MDB_node *leaf, MDB_val *data);

static int	mdb_rebalance(MDB_cursor *mc);
static int	mdb_update_key(MDB_page *mp, indx_t indx, MDB_val *key);
static int	mdb_move_node(MDB_cursor *csrc, MDB_cursor *cdst);
static int	mdb_merge(MDB_cursor *csrc, MDB_cursor *cdst);
static int	mdb_split(MDB_cursor *mc, MDB_val *newkey, MDB_val *newdata,
				pgno_t newpgno);
static MDB_page *mdb_new_page(MDB_cursor *mc, uint32_t flags, int num);

static void	cursor_pop_page(MDB_cursor *mc);
static int	cursor_push_page(MDB_cursor *mc, MDB_page *mp);

static int	mdb_sibling(MDB_cursor *mc, int move_right);
static int	mdb_cursor_next(MDB_cursor *mc, MDB_val *key, MDB_val *data, MDB_cursor_op op);
static int	mdb_cursor_prev(MDB_cursor *mc, MDB_val *key, MDB_val *data, MDB_cursor_op op);
static int	mdb_cursor_set(MDB_cursor *mc, MDB_val *key, MDB_val *data, MDB_cursor_op op,
				int *exactp);
static int	mdb_cursor_first(MDB_cursor *mc, MDB_val *key, MDB_val *data);
static int	mdb_cursor_last(MDB_cursor *mc, MDB_val *key, MDB_val *data);

static void	mdb_xcursor_init0(MDB_cursor *mc);
static void	mdb_xcursor_init1(MDB_cursor *mc, MDB_node *node);
static void	mdb_xcursor_init2(MDB_cursor *mc);
static void	mdb_xcursor_fini(MDB_cursor *mc);

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 void mdb_default_cmp(MDB_txn *txn, MDB_dbi dbi);

/** @cond */
static MDB_cmp_func	memncmp, memnrcmp, intcmp, cintcmp;
/** @endcond */

#ifdef _WIN32
static SECURITY_DESCRIPTOR mdb_null_sd;
static SECURITY_ATTRIBUTES mdb_all_sa;
static int mdb_sec_inited;
#endif

/** Return the library version info. */
char *
mdb_version(int *major, int *minor, int *patch)
{
	if (major) *major = MDB_VERSION_MAJOR;
	if (minor) *minor = MDB_VERSION_MINOR;
	if (patch) *patch = MDB_VERSION_PATCH;
	return MDB_VERSION_STRING;
}

/** Table of descriptions for MDB @ref errors */
static char *const mdb_errstr[] = {
	"MDB_KEYEXIST: Key/data pair already exists",
	"MDB_NOTFOUND: No matching key/data pair found",
	"MDB_PAGE_NOTFOUND: Requested page not found",
	"MDB_CORRUPTED: Located page was wrong type",
	"MDB_PANIC: Update of meta page failed",
	"MDB_VERSION_MISMATCH: Database environment version mismatch"
};

char *
mdb_strerror(int err)
{
	if (!err)
		return ("Successful return: 0");

	if (err >= MDB_KEYEXIST && err <= MDB_VERSION_MISMATCH)
		return mdb_errstr[err - MDB_KEYEXIST];

	return strerror(err);
}

#if DEBUG
/** Display a key in hexadecimal and return the address of the result.
 * @param[in] key the key to display
 * @param[in] buf the buffer to write into. Should always be #DKBUF.
 * @return The key in hexadecimal form.
 */
static char *
mdb_dkey(MDB_val *key, char *buf)
{
	char *ptr = buf;
	unsigned char *c = key->mv_data;
	unsigned int i;
	if (key->mv_size > MAXKEYSIZE)
		return "MAXKEYSIZE";
	/* may want to make this a dynamic check: if the key is mostly
	 * printable characters, print it as-is instead of converting to hex.
	 */
#if 1
	for (i=0; i<key->mv_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)
{
	return txn->mt_dbxs[dbi].md_cmp(a, b);
}

/** Compare two data items according to a particular database.
 * This returns a comparison as if the two items were data items of
 * a sorted duplicates #MDB_DUPSORT database.
 * @param[in] txn A transaction handle returned by #mdb_txn_begin()
 * @param[in] dbi A database handle returned by #mdb_open()
 * @param[in] a The first item to compare
 * @param[in] b The second item to compare
 * @return < 0 if a < b, 0 if a == b, > 0 if a > b
 */
int
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);
	else
		return EINVAL;	/* too bad you can't distinguish this from a valid result */
}

/** Allocate pages for writing.
 * If there are free pages available from older transactions, they
 * will be re-used first. Otherwise a new page will be allocated.
 * @param[in] mc cursor A cursor handle identifying the transaction and
 *	database for which we are allocating.
 * @param[in] num the number of pages to allocate.
 * @return Address of the allocated page(s). Requests for multiple pages
 *  will always be satisfied by a single contiguous chunk of memory.
 */
static MDB_page *
mdb_alloc_page(MDB_cursor *mc, int num)
{
	MDB_txn *txn = mc->mc_txn;
	MDB_page *np;
	pgno_t pgno = P_INVALID;
	ID2 mid;

	if (txn->mt_txnid > 2) {

		if (!txn->mt_env->me_pghead && mc->mc_dbi != FREE_DBI &&
			txn->mt_dbs[FREE_DBI].md_root != P_INVALID) {
			/* See if there's anything in the free DB */
			MDB_cursor m2;
			MDB_node *leaf;
			ULONG *kptr, oldest;

			m2.mc_txn = txn;
			m2.mc_dbi = FREE_DBI;
			m2.mc_snum = 0;
			m2.mc_flags = 0;
			mdb_search_page(&m2, NULL, 0);
			leaf = NODEPTR(m2.mc_pg[m2.mc_top], 0);
			kptr = (ULONG *)NODEKEY(leaf);

			{
				unsigned int i;
				oldest = txn->mt_txnid - 1;
				for (i=0; i<txn->mt_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_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; i<idl[0]; i++) {
						DPRINTF("IDL %lu", idl[i+1]);
					}
				}
#endif
				/* drop this IDL from the DB */
				m2.mc_ki[m2.mc_top] = 0;
				m2.mc_flags = C_INITIALIZED;
				mdb_cursor_del(&m2, 0);
			}
		}
		if (txn->mt_env->me_pghead) {
			MDB_oldpages *mop = txn->mt_env->me_pghead;
			if (num > 1) {
				/* FIXME: For now, always use fresh pages. We
				 * really ought to search the free list for a
				 * contiguous range.
				 */
				;
			} else {
				/* peel pages off tail, so we only have to truncate the list */
				pgno = MDB_IDL_LAST(mop->mo_pages);
				if (MDB_IDL_IS_RANGE(mop->mo_pages)) {
					mop->mo_pages[2]++;
					if (mop->mo_pages[2] > mop->mo_pages[1])
						mop->mo_pages[0] = 0;
				} else {
					mop->mo_pages[0]--;
				}
				if (MDB_IDL_IS_ZERO(mop->mo_pages)) {
					txn->mt_env->me_pghead = mop->mo_next;
					free(mop);
				}
			}
		}
	}

	if (pgno == P_INVALID) {
		/* DB size is maxed out */
		if (txn->mt_next_pgno + num >= txn->mt_env->me_maxpg)
			return NULL;
	}
	if (txn->mt_env->me_dpages && num == 1) {
		np = txn->mt_env->me_dpages;
		txn->mt_env->me_dpages = np->mp_next;
	} else {
		if ((np = malloc(txn->mt_env->me_psize * num )) == NULL)
			return NULL;
	}
	if (pgno == P_INVALID) {
		np->mp_pgno = txn->mt_next_pgno;
		txn->mt_next_pgno += num;
	} else {
		np->mp_pgno = pgno;
	}
	mid.mid = np->mp_pgno;
	mid.mptr = np;
	mdb_mid2l_insert(txn->mt_u.dirty_list, &mid);

	return np;
}

/** Touch a page: make it dirty and re-insert into tree with updated pgno.
 * @param[in] mc cursor pointing to the page to be touched
 * @return 0 on success, non-zero on failure.
 */
static int
mdb_touch(MDB_cursor *mc)
{
	MDB_page *mp = mc->mc_pg[mc->mc_top];
	pgno_t	pgno;

	if (!F_ISSET(mp->mp_flags, P_DIRTY)) {
		MDB_page *np;
		if ((np = mdb_alloc_page(mc, 1)) == NULL)
			return ENOMEM;
		DPRINTF("touched db %u page %lu -> %lu", mc->mc_dbi, mp->mp_pgno, np->mp_pgno);
		assert(mp->mp_pgno != np->mp_pgno);
		mdb_midl_append(mc->mc_txn->mt_free_pgs, mp->mp_pgno);
		pgno = np->mp_pgno;
		memcpy(np, mp, mc->mc_txn->mt_env->me_psize);
		mp = np;
		mp->mp_pgno = pgno;
		mp->mp_flags |= P_DIRTY;

		mc->mc_pg[mc->mc_top] = mp;
		/** If this page has a parent, update the parent to point to
		 * this new page.
		 */
		if (mc->mc_top)
			SETPGNO(NODEPTR(mc->mc_pg[mc->mc_top-1], mc->mc_ki[mc->mc_top-1]), mp->mp_pgno);
	}
	return 0;
}

int
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 = ErrCode();
	}
	return rc;
}

static inline void
mdb_txn_reset0(MDB_txn *txn);

/** Common code for #mdb_txn_begin() and #mdb_txn_renew().
 * @param[in] txn the transaction handle to initialize
 * @return 0 on success, non-zero on failure. This can only
 * fail for read-only transactions, and then only if the
 * reader table is full.
 */
static inline int
mdb_txn_renew0(MDB_txn *txn)
{
	MDB_env *env = txn->mt_env;

	if (txn->mt_flags & MDB_TXN_RDONLY) {
		MDB_reader *r = pthread_getspecific(env->me_txkey);
		if (!r) {
			unsigned int i;
			pid_t pid = getpid();
			pthread_t tid = pthread_self();

			LOCK_MUTEX_R(env);
			for (i=0; i<env->me_txns->mti_numreaders; i++)
				if (env->me_txns->mti_readers[i].mr_pid == 0)
					break;
			if (i == env->me_maxreaders) {
				UNLOCK_MUTEX_R(env);
				return ENOMEM;
			}
			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;
			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 {
		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;
	}

	/* 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_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));
	LAZY_RWLOCK_UNLOCK(&env->me_dblock);

	return MDB_SUCCESS;
}

int
mdb_txn_renew(MDB_txn *txn)
{
	int rc;

	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("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, unsigned int flags, MDB_txn **ret)
{
	MDB_txn *txn;
	int rc;

	if (env->me_flags & MDB_FATAL_ERROR) {
		DPUTS("environment had fatal error, must shutdown!");
		return MDB_PANIC;
	}
	if ((txn = calloc(1, sizeof(MDB_txn) + env->me_maxdbs * sizeof(MDB_db))) == NULL) {
		DPRINTF("calloc: %s", strerror(ErrCode()));
		return ENOMEM;
	}
	txn->mt_dbs = (MDB_db *)(txn+1);
	if (flags & MDB_RDONLY) {