git.sur5r.net Git - openldap/blob - libraries/libmdb/mdb.c

   1 /** @file mdb.c
   2  *      @brief memory-mapped database library
   3  *
   4  *      A Btree-based database management library modeled loosely on the
   5  *      BerkeleyDB API, but much simplified.
   6  */
   7 /*
   8  * Copyright 2011 Howard Chu, Symas Corp.
   9  * All rights reserved.
  10  *
  11  * Redistribution and use in source and binary forms, with or without
  12  * modification, are permitted only as authorized by the OpenLDAP
  13  * Public License.
  14  *
  15  * A copy of this license is available in the file LICENSE in the
  16  * top-level directory of the distribution or, alternatively, at
  17  * <http://www.OpenLDAP.org/license.html>.
  18  *
  19  * This code is derived from btree.c written by Martin Hedenfalk.
  20  *
  21  * Copyright (c) 2009, 2010 Martin Hedenfalk <martin@bzero.se>
  22  *
  23  * Permission to use, copy, modify, and distribute this software for any
  24  * purpose with or without fee is hereby granted, provided that the above
  25  * copyright notice and this permission notice appear in all copies.
  26  *
  27  * THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
  28  * WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
  29  * MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR
  30  * ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES
  31  * WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN
  32  * ACTION OF CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF
  33  * OR IN CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
  34  */
  35 #include <sys/types.h>
  36 #include <sys/stat.h>
  37 #include <sys/param.h>
  38 #ifdef _WIN32
  39 #include <windows.h>
  40 #else
  41 #include <sys/uio.h>
  42 #include <sys/mman.h>
  43 #ifdef HAVE_SYS_FILE_H
  44 #include <sys/file.h>
  45 #endif
  46 #include <fcntl.h>
  47 #endif
  48
  49 #include <assert.h>
  50 #include <errno.h>
  51 #include <stddef.h>
  52 #include <stdint.h>
  53 #include <stdio.h>
  54 #include <stdlib.h>
  55 #include <string.h>
  56 #include <time.h>
  57 #include <unistd.h>
  58
  59 #ifndef _WIN32
  60 #include <pthread.h>
  61 #endif
  62
  63 #include "mdb.h"
  64 #include "midl.h"
  65
  66 /** @defgroup internal  MDB Internals
  67  *      @{
  68  */
  69 /** @defgroup compat    Windows Compatibility Macros
  70  *      @{
  71  */
  72 #ifdef _WIN32
  73 #define pthread_t       DWORD
  74 #define pthread_mutex_t HANDLE
  75 #define pthread_key_t   DWORD
  76 #define pthread_self()  GetCurrentThreadId()
  77 #define pthread_key_create(x,y) *(x) = TlsAlloc()
  78 #define pthread_key_delete(x)   TlsFree(x)
  79 #define pthread_getspecific(x)  TlsGetValue(x)
  80 #define pthread_setspecific(x,y)        TlsSetValue(x,y)
  81 #define pthread_mutex_unlock(x) ReleaseMutex(x)
  82 #define pthread_mutex_lock(x)   WaitForSingleObject(x, INFINITE)
  83 #define LOCK_MUTEX_R(env)       pthread_mutex_lock(env->me_rmutex)
  84 #define UNLOCK_MUTEX_R(env)     pthread_mutex_unlock(env->me_rmutex)
  85 #define LOCK_MUTEX_W(env)       pthread_mutex_lock(env->me_wmutex)
  86 #define UNLOCK_MUTEX_W(env)     pthread_mutex_unlock(env->me_wmutex)
  87 #define getpid()        GetCurrentProcessId()
  88 #define fdatasync(fd)   !FlushFileBuffers(fd)
  89 #define ErrCode()       GetLastError()
  90 #define GetPageSize(x)  {SYSTEM_INFO si; GetSystemInfo(&si); (x) = si.dwPageSize;}
  91 #define close(fd)       CloseHandle(fd)
  92 #define munmap(ptr,len) UnmapViewOfFile(ptr)
  93 #else
  94 #define LOCK_MUTEX_R(env)       pthread_mutex_lock(&env->me_txns->mti_mutex)
  95 #define UNLOCK_MUTEX_R(env)     pthread_mutex_unlock(&env->me_txns->mti_mutex)
  96 #define LOCK_MUTEX_W(env)       pthread_mutex_lock(&env->me_txns->mti_wmutex)
  97 #define UNLOCK_MUTEX_W(env)     pthread_mutex_unlock(&env->me_txns->mti_wmutex)
  98 #define ErrCode()       errno
  99 #define HANDLE  int
 100 #define INVALID_HANDLE_VALUE    -1
 101 #define GetPageSize(x)  (x) = sysconf(_SC_PAGE_SIZE)
 102 #endif
 103
 104 /** @} */
 105
 106 #ifndef _WIN32
 107 /* Note: If O_DSYNC is undefined but exists in /usr/include,
 108  * preferably set some compiler flag to get the definition.
 109  * Otherwise compile with the less efficient -DMDB_DSYNC=O_SYNC.
 110  */
 111 #ifndef MDB_DSYNC
 112 # define MDB_DSYNC      O_DSYNC
 113 #endif
 114 #endif
 115
 116 typedef ULONG           pgno_t;
 117
 118 #ifndef DEBUG
 119 #define DEBUG 0
 120 #endif
 121
 122 #if !(__STDC_VERSION__ >= 199901L || defined(__GNUC__))
 123 # define DPRINTF        (void)  /* Vararg macros may be unsupported */
 124 #elif DEBUG
 125 # define DPRINTF(fmt, ...)      /* Requires 2 or more args */ \
 126         fprintf(stderr, "%s:%d:(%p) " fmt "\n", __func__, __LINE__, pthread_self(), __VA_ARGS__)
 127 #else
 128 # define DPRINTF(fmt, ...)      ((void) 0)
 129 #endif
 130 #define DPUTS(arg)      DPRINTF("%s", arg)
 131
 132 #define PAGESIZE         4096
 133 #define MDB_MINKEYS      2
 134 #define MDB_MAGIC        0xBEEFC0DE
 135 #define MDB_VERSION      1
 136 #define MAXKEYSIZE       511
 137 #if DEBUG
 138 #define KBUF    (MAXKEYSIZE*2+1)
 139 #define DKBUF   char kbuf[KBUF]
 140 #define DKEY(x) mdb_dkey(x, kbuf)
 141 #else
 142 #define DKBUF
 143 #define DKEY(x)
 144 #endif
 145
 146 /* The DB view is always consistent because all writes are wrapped in
 147  * the wmutex. Finer-grained locks aren't necessary.
 148  */
 149 #ifndef LAZY_LOCKS
 150 #define LAZY_LOCKS      1
 151 #endif
 152 #if     LAZY_LOCKS
 153 #define LAZY_MUTEX_LOCK(x)
 154 #define LAZY_MUTEX_UNLOCK(x)
 155 #define LAZY_RWLOCK_UNLOCK(x)
 156 #define LAZY_RWLOCK_WRLOCK(x)
 157 #define LAZY_RWLOCK_RDLOCK(x)
 158 #define LAZY_RWLOCK_DEF(x)
 159 #define LAZY_RWLOCK_INIT(x,y)
 160 #define LAZY_RWLOCK_DESTROY(x)
 161 #else
 162 #define LAZY_MUTEX_LOCK(x)              pthread_mutex_lock(x)
 163 #define LAZY_MUTEX_UNLOCK(x)    pthread_mutex_unlock(x)
 164 #define LAZY_RWLOCK_UNLOCK(x)   pthread_rwlock_unlock(x)
 165 #define LAZY_RWLOCK_WRLOCK(x)   pthread_rwlock_wrlock(x)
 166 #define LAZY_RWLOCK_RDLOCK(x)   pthread_rwlock_rdlock(x)
 167 #define LAZY_RWLOCK_DEF(x)              pthread_rwlock_t        x
 168 #define LAZY_RWLOCK_INIT(x,y)   pthread_rwlock_init(x,y)
 169 #define LAZY_RWLOCK_DESTROY(x)  pthread_rwlock_destroy(x)
 170 #endif
 171
 172 #define P_INVALID        (~0UL)
 173
 174 #define F_ISSET(w, f)    (((w) & (f)) == (f))
 175
 176 typedef uint16_t         indx_t;
 177
 178 #define DEFAULT_READERS 126
 179 #define DEFAULT_MAPSIZE 1048576
 180
 181 /* Lock descriptor stuff */
 182 #ifndef CACHELINE
 183 #define CACHELINE       64      /* most CPUs. Itanium uses 128 */
 184 #endif
 185
 186 typedef struct MDB_rxbody {
 187         ULONG           mrb_txnid;
 188         pid_t           mrb_pid;
 189         pthread_t       mrb_tid;
 190 } MDB_rxbody;
 191
 192 typedef struct MDB_reader {
 193         union {
 194                 MDB_rxbody mrx;
 195 #define mr_txnid        mru.mrx.mrb_txnid
 196 #define mr_pid  mru.mrx.mrb_pid
 197 #define mr_tid  mru.mrx.mrb_tid
 198                 /* cache line alignment */
 199                 char pad[(sizeof(MDB_rxbody)+CACHELINE-1) & ~(CACHELINE-1)];
 200         } mru;
 201 } MDB_reader;
 202
 203 typedef struct MDB_txbody {
 204         uint32_t        mtb_magic;
 205         uint32_t        mtb_version;
 206 /* For POSIX the actual mutexes reside in shared memory.
 207  * On Windows, mutexes are allocated by the kernel; we store
 208  * the name in shared memory so that other processes can
 209  * grab them.
 210  */
 211 #ifdef _WIN32
 212         char    mtb_rmname[32];
 213 #else
 214         pthread_mutex_t mtb_mutex;
 215 #endif
 216         ULONG           mtb_txnid;
 217         uint32_t        mtb_numreaders;
 218         uint32_t        mtb_me_toggle;
 219 } MDB_txbody;
 220
 221 typedef struct MDB_txninfo {
 222         union {
 223                 MDB_txbody mtb;
 224 #define mti_magic       mt1.mtb.mtb_magic
 225 #define mti_version     mt1.mtb.mtb_version
 226 #define mti_mutex       mt1.mtb.mtb_mutex
 227 #define mti_rmname      mt1.mtb.mtb_rmname
 228 #define mti_txnid       mt1.mtb.mtb_txnid
 229 #define mti_numreaders  mt1.mtb.mtb_numreaders
 230 #define mti_me_toggle   mt1.mtb.mtb_me_toggle
 231                 char pad[(sizeof(MDB_txbody)+CACHELINE-1) & ~(CACHELINE-1)];
 232         } mt1;
 233         union {
 234 #ifdef _WIN32
 235                 char mt2_wmname[32];
 236 #define mti_wmname      mt2.mt2_wmname
 237 #else
 238                 pthread_mutex_t mt2_wmutex;
 239 #define mti_wmutex      mt2.mt2_wmutex
 240 #endif
 241                 char pad[(sizeof(pthread_mutex_t)+CACHELINE-1) & ~(CACHELINE-1)];
 242         } mt2;
 243         MDB_reader      mti_readers[1];
 244 } MDB_txninfo;
 245
 246 /* Common header for all page types. Overflow pages
 247  * occupy a number of contiguous pages with no
 248  * headers on any page after the first.
 249  */
 250 typedef struct MDB_page {               /* represents a page of storage */
 251 #define mp_pgno         mp_p.p_pgno
 252 #define mp_next         mp_p.p_align
 253         union padded {
 254                 pgno_t          p_pgno;         /* page number */
 255                 void *          p_align;        /* for IL32P64 */
 256         } mp_p;
 257 #define P_BRANCH         0x01           /* branch page */
 258 #define P_LEAF           0x02           /* leaf page */
 259 #define P_OVERFLOW       0x04           /* overflow page */
 260 #define P_META           0x08           /* meta page */
 261 #define P_DIRTY          0x10           /* dirty page */
 262 #define P_LEAF2          0x20           /* DB with small, fixed size keys and no data */
 263         uint32_t        mp_flags;
 264 #define mp_lower        mp_pb.pb.pb_lower
 265 #define mp_upper        mp_pb.pb.pb_upper
 266 #define mp_pages        mp_pb.pb_pages
 267         union page_bounds {
 268                 struct {
 269                         indx_t          pb_lower;               /* lower bound of free space */
 270                         indx_t          pb_upper;               /* upper bound of free space */
 271                 } pb;
 272                 uint32_t        pb_pages;       /* number of overflow pages */
 273         } mp_pb;
 274         indx_t          mp_ptrs[1];             /* dynamic size */
 275 } MDB_page;
 276
 277 #define PAGEHDRSZ        ((unsigned) offsetof(MDB_page, mp_ptrs))
 278
 279 #define NUMKEYS(p)       (((p)->mp_lower - PAGEHDRSZ) >> 1)
 280 #define SIZELEFT(p)      (indx_t)((p)->mp_upper - (p)->mp_lower)
 281 #define PAGEFILL(env, p) (1000L * ((env)->me_psize - PAGEHDRSZ - SIZELEFT(p)) / \
 282                                 ((env)->me_psize - PAGEHDRSZ))
 283 #define IS_LEAF(p)       F_ISSET((p)->mp_flags, P_LEAF)
 284 #define IS_LEAF2(p)      F_ISSET((p)->mp_flags, P_LEAF2)
 285 #define IS_BRANCH(p)     F_ISSET((p)->mp_flags, P_BRANCH)
 286 #define IS_OVERFLOW(p)   F_ISSET((p)->mp_flags, P_OVERFLOW)
 287
 288 #define OVPAGES(size, psize)    ((PAGEHDRSZ-1 + (size)) / (psize) + 1)
 289
 290 typedef struct MDB_db {
 291         uint32_t        md_pad;         /* also ksize for LEAF2 pages */
 292         uint16_t        md_flags;
 293         uint16_t        md_depth;
 294         ULONG           md_branch_pages;
 295         ULONG           md_leaf_pages;
 296         ULONG           md_overflow_pages;
 297         ULONG           md_entries;
 298         pgno_t          md_root;
 299 } MDB_db;
 300
 301 #define FREE_DBI        0
 302 #define MAIN_DBI        1
 303
 304 typedef struct MDB_meta {                       /* meta (footer) page content */
 305         uint32_t        mm_magic;
 306         uint32_t        mm_version;
 307         void            *mm_address;            /* address for fixed mapping */
 308         size_t          mm_mapsize;                     /* size of mmap region */
 309         MDB_db          mm_dbs[2];                      /* first is free space, 2nd is main db */
 310 #define mm_psize        mm_dbs[0].md_pad
 311 #define mm_flags        mm_dbs[0].md_flags
 312         pgno_t          mm_last_pg;                     /* last used page in file */
 313         ULONG           mm_txnid;                       /* txnid that committed this page */
 314 } MDB_meta;
 315
 316 typedef struct MDB_oldpages {
 317         struct MDB_oldpages *mo_next;
 318         ULONG           mo_txnid;
 319         pgno_t          mo_pages[1];    /* dynamic */
 320 } MDB_oldpages;
 321
 322 static MDB_page *mdb_alloc_page(MDB_cursor *mc, int num);
 323 static int              mdb_touch(MDB_cursor *mc);
 324
 325 /* Enough space for 2^32 nodes with minimum of 2 keys per node. I.e., plenty.
 326  * At 4 keys per node, enough for 2^64 nodes, so there's probably no need to
 327  * raise this on a 64 bit machine.
 328  */
 329 #define CURSOR_STACK             32
 330
 331 struct MDB_xcursor;
 332
 333 struct MDB_cursor {
 334         struct MDB_xcursor      *mc_xcursor;
 335         MDB_txn         *mc_txn;
 336         MDB_dbi         mc_dbi;
 337         unsigned short  mc_snum;                /* number of pushed pages */
 338         unsigned short  mc_top;         /* index of top page, mc_snum-1 */
 339         unsigned int    mc_flags;
 340 #define C_INITIALIZED   0x01
 341 #define C_EOF   0x02
 342 #define C_XDIRTY        0x04
 343         MDB_page        *mc_pg[CURSOR_STACK];   /* stack of pushed pages */
 344         indx_t          mc_ki[CURSOR_STACK];    /* stack of page indices */
 345 };
 346
 347 #define METADATA(p)      ((void *)((char *)(p) + PAGEHDRSZ))
 348
 349 /* We guarantee 2-byte alignment for nodes */
 350 typedef struct MDB_node {
 351         /* lo and hi are used for data size on leaf nodes and for
 352          * child pgno on branch nodes. On 64 bit platforms, flags
 353          * is also used for pgno. (branch nodes ignore flags)
 354          */
 355         unsigned short  mn_lo;
 356         unsigned short  mn_hi;
 357         unsigned short  mn_flags;
 358 #define F_BIGDATA        0x01                   /* data put on overflow page */
 359 #define F_SUBDATA        0x02                   /* data is a sub-database */
 360 #define F_DUPDATA        0x04                   /* data has duplicates */
 361         unsigned short  mn_ksize;                               /* key size */
 362         char            mn_data[1];
 363 } MDB_node;
 364
 365 typedef struct MDB_dbx {
 366         MDB_val         md_name;
 367         MDB_cmp_func    *md_cmp;                /* user compare function */
 368         MDB_cmp_func    *md_dcmp;               /* user dupsort function */
 369         MDB_rel_func    *md_rel;                /* user relocate function */
 370         MDB_dbi md_parent;
 371         unsigned int    md_dirty;
 372 } MDB_dbx;
 373
 374 struct MDB_txn {
 375         pgno_t          mt_next_pgno;   /* next unallocated page */
 376         ULONG           mt_txnid;
 377         MDB_env         *mt_env;
 378         pgno_t          *mt_free_pgs;   /* this is an IDL */
 379         union {
 380                 ID2L    dirty_list;     /* modified pages */
 381                 MDB_reader      *reader;
 382         } mt_u;
 383         MDB_dbx         *mt_dbxs;               /* array */
 384         MDB_db          *mt_dbs;
 385         unsigned int    mt_numdbs;
 386
 387 #define MDB_TXN_RDONLY          0x01            /* read-only transaction */
 388 #define MDB_TXN_ERROR           0x02            /* an error has occurred */
 389         unsigned int    mt_flags;
 390         unsigned int    mt_toggle;
 391 };
 392
 393 /* Context for sorted-dup records */
 394 typedef struct MDB_xcursor {
 395         MDB_cursor mx_cursor;
 396         MDB_txn mx_txn;
 397         MDB_dbx mx_dbxs[4];
 398         MDB_db  mx_dbs[4];
 399 } MDB_xcursor;
 400
 401 struct MDB_env {
 402         HANDLE          me_fd;
 403         HANDLE          me_lfd;
 404         HANDLE          me_mfd;                 /* just for writing the meta pages */
 405 #define MDB_FATAL_ERROR 0x80000000U
 406         uint32_t        me_flags;
 407         uint32_t        me_extrapad;    /* unused for now */
 408         unsigned int    me_maxreaders;
 409         unsigned int    me_numdbs;
 410         unsigned int    me_maxdbs;
 411         char            *me_path;
 412         char            *me_map;
 413         MDB_txninfo     *me_txns;
 414         MDB_meta        *me_metas[2];
 415         MDB_txn         *me_txn;                /* current write transaction */
 416         size_t          me_mapsize;
 417         off_t           me_size;                /* current file size */
 418         pgno_t          me_maxpg;               /* me_mapsize / me_psize */
 419         unsigned int    me_psize;
 420         unsigned int    me_db_toggle;
 421         MDB_dbx         *me_dbxs;               /* array */
 422         MDB_db          *me_dbs[2];
 423         MDB_oldpages *me_pghead;
 424         pthread_key_t   me_txkey;       /* thread-key for readers */
 425         MDB_page        *me_dpages;
 426         pgno_t          me_free_pgs[MDB_IDL_UM_SIZE];
 427         ID2                     me_dirty_list[MDB_IDL_UM_SIZE];
 428         LAZY_RWLOCK_DEF(me_dblock);
 429 #ifdef _WIN32
 430         HANDLE          me_rmutex;              /* Windows mutexes don't reside in shared mem */
 431         HANDLE          me_wmutex;
 432 #endif
 433 };
 434
 435 #define NODESIZE         offsetof(MDB_node, mn_data)
 436
 437 #define INDXSIZE(k)      (NODESIZE + ((k) == NULL ? 0 : (k)->mv_size))
 438 #define LEAFSIZE(k, d)   (NODESIZE + (k)->mv_size + (d)->mv_size)
 439 #define NODEPTR(p, i)    ((MDB_node *)((char *)(p) + (p)->mp_ptrs[i]))
 440 #define NODEKEY(node)    (void *)((node)->mn_data)
 441 #define NODEDATA(node)   (void *)((char *)(node)->mn_data + (node)->mn_ksize)
 442 #if LONG_MAX == 0x7fffffff
 443 #define NODEPGNO(node)   ((node)->mn_lo | ((node)->mn_hi << 16))
 444 #define SETPGNO(node,pgno)      do { \
 445         (node)->mn_lo = (pgno) & 0xffff; (node)->mn_hi = (pgno) >> 16;} while(0)
 446 #else
 447 #define NODEPGNO(node)   ((node)->mn_lo | ((node)->mn_hi << 16) | ((unsigned long)(node)->mn_flags << 32))
 448 #define SETPGNO(node,pgno)      do { \
 449         (node)->mn_lo = (pgno) & 0xffff; (node)->mn_hi = (pgno) >> 16; \
 450         (node)->mn_flags = (pgno) >> 32; } while(0)
 451 #endif
 452 #define NODEDSZ(node)    ((node)->mn_lo | ((unsigned)(node)->mn_hi << 16))
 453 #define SETDSZ(node,size)       do { \
 454         (node)->mn_lo = (size) & 0xffff; (node)->mn_hi = (size) >> 16;} while(0)
 455 #define NODEKSZ(node)    ((node)->mn_ksize)
 456 #define LEAF2KEY(p, i, ks)      ((char *)(p) + PAGEHDRSZ + ((i)*(ks)))
 457
 458 #define MDB_SET_KEY(node, key)  if (key!=NULL) {(key)->mv_size = NODEKSZ(node); (key)->mv_data = NODEKEY(node);}
 459
 460 #define MDB_COMMIT_PAGES         64     /* max number of pages to write in one commit */
 461
 462 static int  mdb_search_page_root(MDB_cursor *mc,
 463                             MDB_val *key, int modify);
 464 static int  mdb_search_page(MDB_cursor *mc,
 465                             MDB_val *key, int modify);
 466
 467 static int  mdb_env_read_header(MDB_env *env, MDB_meta *meta);
 468 static int  mdb_env_read_meta(MDB_env *env, int *which);
 469 static int  mdb_env_write_meta(MDB_txn *txn);
 470 static int  mdb_get_page(MDB_txn *txn, pgno_t pgno, MDB_page **mp);
 471
 472 static MDB_node *mdb_search_node(MDB_cursor *mc, MDB_val *key, int *exactp);
 473 static int  mdb_add_node(MDB_cursor *mc, indx_t indx,
 474                             MDB_val *key, MDB_val *data, pgno_t pgno, uint8_t flags);
 475 static void mdb_del_node(MDB_page *mp, indx_t indx, int ksize);
 476 static int mdb_del0(MDB_cursor *mc, MDB_node *leaf);
 477 static int  mdb_read_data(MDB_txn *txn, MDB_node *leaf, MDB_val *data);
 478
 479 static int      mdb_rebalance(MDB_cursor *mc);
 480 static int      mdb_update_key(MDB_page *mp, indx_t indx, MDB_val *key);
 481 static int      mdb_move_node(MDB_cursor *csrcrc, MDB_cursor *cdstst);
 482 static int      mdb_merge(MDB_cursor *csrcrc, MDB_cursor *cdstst);
 483 static int      mdb_split(MDB_cursor *mc, MDB_val *newkey, MDB_val *newdata,
 484                                 pgno_t newpgno);
 485 static MDB_page *mdb_new_page(MDB_cursor *mc, uint32_t flags, int num);
 486
 487 static void     cursor_pop_page(MDB_cursor *mc);
 488 static int      cursor_push_page(MDB_cursor *mc, MDB_page *mp);
 489
 490 static int      mdb_sibling(MDB_cursor *mc, int move_right);
 491 static int      mdb_cursor_next(MDB_cursor *mc, MDB_val *key, MDB_val *data, MDB_cursor_op op);
 492 static int      mdb_cursor_prev(MDB_cursor *mc, MDB_val *key, MDB_val *data, MDB_cursor_op op);
 493 static int      mdb_cursor_set(MDB_cursor *mc, MDB_val *key, MDB_val *data, MDB_cursor_op op,
 494                                 int *exactp);
 495 static int      mdb_cursor_first(MDB_cursor *mc, MDB_val *key, MDB_val *data);
 496 static int      mdb_cursor_last(MDB_cursor *mc, MDB_val *key, MDB_val *data);
 497
 498 static void     mdb_xcursor_init0(MDB_cursor *mc);
 499 static void     mdb_xcursor_init1(MDB_cursor *mc, MDB_node *node);
 500 static void     mdb_xcursor_init2(MDB_cursor *mc);
 501 static void     mdb_xcursor_fini(MDB_cursor *mc);
 502
 503 static size_t   mdb_leaf_size(MDB_env *env, MDB_val *key, MDB_val *data);
 504 static size_t   mdb_branch_size(MDB_env *env, MDB_val *key);
 505
 506 static void mdb_default_cmp(MDB_txn *txn, MDB_dbi dbi);
 507
 508 static MDB_cmp_func     memncmp, memnrcmp, intcmp, cintcmp;
 509
 510 #ifdef _WIN32
 511 static SECURITY_DESCRIPTOR mdb_null_sd;
 512 static SECURITY_ATTRIBUTES mdb_all_sa;
 513 static int mdb_sec_inited;
 514 #endif
 515
 516 char *
 517 mdb_version(int *maj, int *min, int *pat)
 518 {
 519         if (maj) *maj = MDB_VERSION_MAJOR;
 520         if (min) *min = MDB_VERSION_MINOR;
 521         if (pat) *pat = MDB_VERSION_PATCH;
 522         return MDB_VERSION_STRING;
 523 }
 524
 525 static char *const mdb_errstr[] = {
 526         "MDB_KEYEXIST: Key/data pair already exists",
 527         "MDB_NOTFOUND: No matching key/data pair found",
 528         "MDB_PAGE_NOTFOUND: Requested page not found",
 529         "MDB_CORRUPTED: Located page was wrong type",
 530         "MDB_PANIC: Update of meta page failed",
 531         "MDB_VERSION_MISMATCH: Database environment version mismatch"
 532 };
 533
 534 char *
 535 mdb_strerror(int err)
 536 {
 537         if (!err)
 538                 return ("Successful return: 0");
 539
 540         if (err >= MDB_KEYEXIST && err <= MDB_VERSION_MISMATCH)
 541                 return mdb_errstr[err - MDB_KEYEXIST];
 542
 543         return strerror(err);
 544 }
 545
 546 #if DEBUG
 547 static char *
 548 mdb_dkey(MDB_val *key, char *buf)
 549 {
 550         char *ptr = buf;
 551         unsigned char *c = key->mv_data;
 552         unsigned int i;
 553         if (key->mv_size > MAXKEYSIZE)
 554                 return "MAXKEYSIZE";
 555 #if 1
 556         for (i=0; i<key->mv_size; i++)
 557                 ptr += sprintf(ptr, "%02x", *c++);
 558 #else
 559         sprintf(buf, "%.*s", key->mv_size, key->mv_data);
 560 #endif
 561         return buf;
 562 }
 563 #endif
 564
 565 int
 566 mdb_cmp(MDB_txn *txn, MDB_dbi dbi, const MDB_val *a, const MDB_val *b)
 567 {
 568         return txn->mt_dbxs[dbi].md_cmp(a, b);
 569 }
 570
 571 int
 572 mdb_dcmp(MDB_txn *txn, MDB_dbi dbi, const MDB_val *a, const MDB_val *b)
 573 {
 574         if (txn->mt_dbxs[dbi].md_dcmp)
 575                 return txn->mt_dbxs[dbi].md_dcmp(a, b);
 576         else
 577                 return EINVAL;  /* too bad you can't distinguish this from a valid result */
 578 }
 579
 580 /* Allocate new page(s) for writing */
 581 static MDB_page *
 582 mdb_alloc_page(MDB_cursor *mc, int num)
 583 {
 584         MDB_txn *txn = mc->mc_txn;
 585         MDB_page *np;
 586         pgno_t pgno = P_INVALID;
 587         ID2 mid;
 588
 589         if (txn->mt_txnid > 2) {
 590
 591                 if (!txn->mt_env->me_pghead && mc->mc_dbi != FREE_DBI &&
 592                         txn->mt_dbs[FREE_DBI].md_root != P_INVALID) {
 593                         /* See if there's anything in the free DB */
 594                         MDB_cursor m2;
 595                         MDB_node *leaf;
 596                         ULONG *kptr, oldest;
 597
 598                         m2.mc_txn = txn;
 599                         m2.mc_dbi = FREE_DBI;
 600                         m2.mc_snum = 0;
 601                         m2.mc_flags = 0;
 602                         mdb_search_page(&m2, NULL, 0);
 603                         leaf = NODEPTR(m2.mc_pg[m2.mc_top], 0);
 604                         kptr = (ULONG *)NODEKEY(leaf);
 605
 606                         {
 607                                 unsigned int i;
 608                                 oldest = txn->mt_txnid - 1;
 609                                 for (i=0; i<txn->mt_env->me_txns->mti_numreaders; i++) {
 610                                         ULONG mr = txn->mt_env->me_txns->mti_readers[i].mr_txnid;
 611                                         if (mr && mr < oldest)
 612                                                 oldest = mr;
 613                                 }
 614                         }
 615
 616                         if (oldest > *kptr) {
 617                                 /* It's usable, grab it.
 618                                  */
 619                                 MDB_oldpages *mop;
 620                                 MDB_val data;
 621                                 pgno_t *idl;
 622
 623                                 mdb_read_data(txn, leaf, &data);
 624                                 idl = (ULONG *)data.mv_data;
 625                                 mop = malloc(sizeof(MDB_oldpages) + MDB_IDL_SIZEOF(idl) - sizeof(pgno_t));
 626                                 mop->mo_next = txn->mt_env->me_pghead;
 627                                 mop->mo_txnid = *kptr;
 628                                 txn->mt_env->me_pghead = mop;
 629                                 memcpy(mop->mo_pages, idl, MDB_IDL_SIZEOF(idl));
 630
 631 #if DEBUG > 1
 632                                 {
 633                                         unsigned int i;
 634                                         DPRINTF("IDL read txn %lu root %lu num %lu",
 635                                                 mop->mo_txnid, txn->mt_dbs[FREE_DBI].md_root, idl[0]);
 636                                         for (i=0; i<idl[0]; i++) {
 637                                                 DPRINTF("IDL %lu", idl[i+1]);
 638                                         }
 639                                 }
 640 #endif
 641                                 /* drop this IDL from the DB */
 642                                 m2.mc_ki[m2.mc_top] = 0;
 643                                 m2.mc_flags = C_INITIALIZED;
 644                                 mdb_cursor_del(&m2, 0);
 645                         }
 646                 }
 647                 if (txn->mt_env->me_pghead) {
 648                         MDB_oldpages *mop = txn->mt_env->me_pghead;
 649                         if (num > 1) {
 650                                 /* FIXME: For now, always use fresh pages. We
 651                                  * really ought to search the free list for a
 652                                  * contiguous range.
 653                                  */
 654                                 ;
 655                         } else {
 656                                 /* peel pages off tail, so we only have to truncate the list */
 657                                 pgno = MDB_IDL_LAST(mop->mo_pages);
 658                                 if (MDB_IDL_IS_RANGE(mop->mo_pages)) {
 659                                         mop->mo_pages[2]++;
 660                                         if (mop->mo_pages[2] > mop->mo_pages[1])
 661                                                 mop->mo_pages[0] = 0;
 662                                 } else {
 663                                         mop->mo_pages[0]--;
 664                                 }
 665                                 if (MDB_IDL_IS_ZERO(mop->mo_pages)) {
 666                                         txn->mt_env->me_pghead = mop->mo_next;
 667                                         free(mop);
 668                                 }
 669                         }
 670                 }
 671         }
 672
 673         if (pgno == P_INVALID) {
 674                 /* DB size is maxed out */
 675                 if (txn->mt_next_pgno + num >= txn->mt_env->me_maxpg)
 676                         return NULL;
 677         }
 678         if (txn->mt_env->me_dpages && num == 1) {
 679                 np = txn->mt_env->me_dpages;
 680                 txn->mt_env->me_dpages = np->mp_next;
 681         } else {
 682                 if ((np = malloc(txn->mt_env->me_psize * num )) == NULL)
 683                         return NULL;
 684         }
 685         if (pgno == P_INVALID) {
 686                 np->mp_pgno = txn->mt_next_pgno;
 687                 txn->mt_next_pgno += num;
 688         } else {
 689                 np->mp_pgno = pgno;
 690         }
 691         mid.mid = np->mp_pgno;
 692         mid.mptr = np;
 693         mdb_mid2l_insert(txn->mt_u.dirty_list, &mid);
 694
 695         return np;
 696 }
 697
 698 /* Touch a page: make it dirty and re-insert into tree with updated pgno.
 699  */
 700 static int
 701 mdb_touch(MDB_cursor *mc)
 702 {
 703         MDB_page *mp = mc->mc_pg[mc->mc_top];
 704         pgno_t  pgno;
 705
 706         if (!F_ISSET(mp->mp_flags, P_DIRTY)) {
 707                 MDB_page *np;
 708                 if ((np = mdb_alloc_page(mc, 1)) == NULL)
 709                         return ENOMEM;
 710                 DPRINTF("touched db %u page %lu -> %lu", mc->mc_dbi, mp->mp_pgno, np->mp_pgno);
 711                 assert(mp->mp_pgno != np->mp_pgno);
 712                 mdb_midl_append(mc->mc_txn->mt_free_pgs, mp->mp_pgno);
 713                 pgno = np->mp_pgno;
 714                 memcpy(np, mp, mc->mc_txn->mt_env->me_psize);
 715                 mp = np;
 716                 mp->mp_pgno = pgno;
 717                 mp->mp_flags |= P_DIRTY;
 718
 719                 mc->mc_pg[mc->mc_top] = mp;
 720                 /* Update the page number to new touched page. */
 721                 if (mc->mc_top)
 722                         SETPGNO(NODEPTR(mc->mc_pg[mc->mc_top-1], mc->mc_ki[mc->mc_top-1]), mp->mp_pgno);
 723         }
 724         return 0;
 725 }
 726
 727 int
 728 mdb_env_sync(MDB_env *env, int force)
 729 {
 730         int rc = 0;
 731         if (force || !F_ISSET(env->me_flags, MDB_NOSYNC)) {
 732                 if (fdatasync(env->me_fd))
 733                         rc = ErrCode();
 734         }
 735         return rc;
 736 }
 737
 738 static inline void
 739 mdb_txn_reset0(MDB_txn *txn);
 740
 741 static inline int
 742 mdb_txn_renew0(MDB_txn *txn)
 743 {
 744         MDB_env *env = txn->mt_env;
 745
 746         if (txn->mt_flags & MDB_TXN_RDONLY) {
 747                 MDB_reader *r = pthread_getspecific(env->me_txkey);
 748                 if (!r) {
 749                         unsigned int i;
 750                         pid_t pid = getpid();
 751                         pthread_t tid = pthread_self();
 752
 753                         LOCK_MUTEX_R(env);
 754                         for (i=0; i<env->me_txns->mti_numreaders; i++)
 755                                 if (env->me_txns->mti_readers[i].mr_pid == 0)
 756                                         break;
 757                         if (i == env->me_maxreaders) {
 758                                 UNLOCK_MUTEX_R(env);
 759                                 return ENOMEM;
 760                         }
 761                         env->me_txns->mti_readers[i].mr_pid = pid;
 762                         env->me_txns->mti_readers[i].mr_tid = tid;
 763                         if (i >= env->me_txns->mti_numreaders)
 764                                 env->me_txns->mti_numreaders = i+1;
 765                         UNLOCK_MUTEX_R(env);
 766                         r = &env->me_txns->mti_readers[i];
 767                         pthread_setspecific(env->me_txkey, r);
 768                 }
 769                 txn->mt_txnid = env->me_txns->mti_txnid;
 770                 txn->mt_toggle = env->me_txns->mti_me_toggle;
 771                 r->mr_txnid = txn->mt_txnid;
 772                 txn->mt_u.reader = r;
 773         } else {
 774                 LOCK_MUTEX_W(env);
 775
 776                 txn->mt_txnid = env->me_txns->mti_txnid+1;
 777                 txn->mt_toggle = env->me_txns->mti_me_toggle;
 778                 txn->mt_u.dirty_list = env->me_dirty_list;
 779                 txn->mt_u.dirty_list[0].mid = 0;
 780                 txn->mt_free_pgs = env->me_free_pgs;
 781                 txn->mt_free_pgs[0] = 0;
 782                 txn->mt_next_pgno = env->me_metas[txn->mt_toggle]->mm_last_pg+1;
 783                 env->me_txn = txn;
 784         }
 785
 786         /* Copy the DB arrays */
 787         LAZY_RWLOCK_RDLOCK(&env->me_dblock);
 788         txn->mt_numdbs = env->me_numdbs;
 789         txn->mt_dbxs = env->me_dbxs;    /* mostly static anyway */
 790         memcpy(txn->mt_dbs, env->me_metas[txn->mt_toggle]->mm_dbs, 2 * sizeof(MDB_db));
 791         if (txn->mt_numdbs > 2)
 792                 memcpy(txn->mt_dbs+2, env->me_dbs[env->me_db_toggle]+2,
 793                         (txn->mt_numdbs - 2) * sizeof(MDB_db));
 794         LAZY_RWLOCK_UNLOCK(&env->me_dblock);
 795
 796         return MDB_SUCCESS;
 797 }
 798
 799 int
 800 mdb_txn_renew(MDB_txn *txn)
 801 {
 802         int rc;
 803
 804         if (!txn)
 805                 return EINVAL;
 806
 807         if (txn->mt_env->me_flags & MDB_FATAL_ERROR) {
 808                 DPUTS("environment had fatal error, must shutdown!");
 809                 return MDB_PANIC;
 810         }
 811
 812         rc = mdb_txn_renew0(txn);
 813         if (rc == MDB_SUCCESS) {
 814                 DPRINTF("renew txn %lu%c %p on mdbenv %p, root page %lu",
 815                         txn->mt_txnid, (txn->mt_flags & MDB_TXN_RDONLY) ? 'r' : 'w', txn,
 816                         (void *)txn->mt_env, txn->mt_dbs[MAIN_DBI].md_root);
 817         }
 818         return rc;
 819 }
 820
 821 int
 822 mdb_txn_begin(MDB_env *env, unsigned int flags, MDB_txn **ret)
 823 {
 824         MDB_txn *txn;
 825         int rc;
 826
 827         if (env->me_flags & MDB_FATAL_ERROR) {
 828                 DPUTS("environment had fatal error, must shutdown!");
 829                 return MDB_PANIC;
 830         }
 831         if ((txn = calloc(1, sizeof(MDB_txn) + env->me_maxdbs * sizeof(MDB_db))) == NULL) {
 832                 DPRINTF("calloc: %s", strerror(ErrCode()));
 833                 return ENOMEM;
 834         }
 835         txn->mt_dbs = (MDB_db *)(txn+1);
 836         if (flags & MDB_RDONLY) {
 837                 txn->mt_flags |= MDB_TXN_RDONLY;
 838         }
 839         txn->mt_env = env;
 840
 841         rc = mdb_txn_renew0(txn);
 842         if (rc)
 843                 free(txn);
 844         else {
 845                 *ret = txn;
 846                 DPRINTF("begin txn %lu%c %p on mdbenv %p, root page %lu",
 847                         txn->mt_txnid, (txn->mt_flags & MDB_TXN_RDONLY) ? 'r' : 'w', txn,
 848                         (void *) env, txn->mt_dbs[MAIN_DBI].md_root);
 849         }
 850
 851         return rc;
 852 }
 853
 854 static inline void
 855 mdb_txn_reset0(MDB_txn *txn)
 856 {
 857         MDB_env *env = txn->mt_env;
 858
 859         if (F_ISSET(txn->mt_flags, MDB_TXN_RDONLY)) {
 860                 txn->mt_u.reader->mr_txnid = 0;
 861         } else {
 862                 MDB_oldpages *mop;
 863                 MDB_page *dp;
 864                 unsigned int i;
 865
 866                 /* return all dirty pages to dpage list */
 867                 for (i=1; i<=txn->mt_u.dirty_list[0].mid; i++) {
 868                         dp = txn->mt_u.dirty_list[i].mptr;
 869                         if (!IS_OVERFLOW(dp) || dp->mp_pages == 1) {
 870                                 dp->mp_next = txn->mt_env->me_dpages;
 871                                 txn->mt_env->me_dpages = dp;
 872                         } else {
 873                                 /* large pages just get freed directly */
 874                                 free(dp);
 875                         }
 876                 }
 877
 878                 while ((mop = txn->mt_env->me_pghead)) {
 879                         txn->mt_env->me_pghead = mop->mo_next;
 880                         free(mop);
 881                 }
 882
 883                 env->me_txn = NULL;
 884                 for (i=2; i<env->me_numdbs; i++)
 885                         env->me_dbxs[i].md_dirty = 0;
 886                 UNLOCK_MUTEX_W(env);
 887         }
 888 }
 889
 890 void
 891 mdb_txn_reset(MDB_txn *txn)
 892 {
 893         if (txn == NULL)
 894                 return;
 895
 896         DPRINTF("reset txn %lu%c %p on mdbenv %p, root page %lu",
 897                 txn->mt_txnid, (txn->mt_flags & MDB_TXN_RDONLY) ? 'r' : 'w', txn,
 898                 (void *)txn->mt_env, txn->mt_dbs[MAIN_DBI].md_root);
 899
 900         mdb_txn_reset0(txn);
 901 }
 902
 903 void
 904 mdb_txn_abort(MDB_txn *txn)
 905 {
 906         if (txn == NULL)
 907                 return;
 908
 909         DPRINTF("abort txn %lu%c %p on mdbenv %p, root page %lu",
 910                 txn->mt_txnid, (txn->mt_flags & MDB_TXN_RDONLY) ? 'r' : 'w', txn,
 911                 (void *)txn->mt_env, txn->mt_dbs[MAIN_DBI].md_root);
 912
 913         mdb_txn_reset0(txn);
 914         free(txn);
 915 }
 916
 917 int
 918 mdb_txn_commit(MDB_txn *txn)
 919 {
 920         int              n, done;
 921         unsigned int i;
 922         ssize_t          rc;
 923         off_t            size;
 924         MDB_page        *dp;
 925         MDB_env *env;
 926         pgno_t  next;
 927         MDB_cursor mc;
 928
 929         assert(txn != NULL);
 930         assert(txn->mt_env != NULL);
 931
 932         env = txn->mt_env;
 933
 934         if (F_ISSET(txn->mt_flags, MDB_TXN_RDONLY)) {
 935                 mdb_txn_abort(txn);
 936                 return MDB_SUCCESS;
 937         }
 938
 939         if (txn != env->me_txn) {
 940                 DPUTS("attempt to commit unknown transaction");
 941                 mdb_txn_abort(txn);
 942                 return EINVAL;
 943         }
 944
 945         if (F_ISSET(txn->mt_flags, MDB_TXN_ERROR)) {
 946                 DPUTS("error flag is set, can't commit");
 947                 mdb_txn_abort(txn);
 948                 return EINVAL;
 949         }
 950
 951         if (!txn->mt_u.dirty_list[0].mid)
 952                 goto done;
 953
 954         DPRINTF("committing txn %lu %p on mdbenv %p, root page %lu",
 955             txn->mt_txnid, txn, (void *)env, txn->mt_dbs[MAIN_DBI].md_root);
 956
 957         mc.mc_txn = txn;
 958         mc.mc_dbi = FREE_DBI;
 959         mc.mc_flags = 0;
 960
 961         /* should only be one record now */
 962         if (env->me_pghead) {
 963                 /* make sure first page of freeDB is touched and on freelist */
 964                 mdb_search_page(&mc, NULL, 1);
 965         }
 966         /* save to free list */
 967         if (!MDB_IDL_IS_ZERO(txn->mt_free_pgs)) {
 968                 MDB_val key, data;
 969                 ULONG i;
 970
 971                 /* make sure last page of freeDB is touched and on freelist */
 972                 key.mv_size = MAXKEYSIZE+1;
 973                 key.mv_data = NULL;
 974                 mdb_search_page(&mc, &key, 1);
 975
 976                 mdb_midl_sort(txn->mt_free_pgs);
 977 #if DEBUG > 1
 978                 {
 979                         unsigned int i;
 980                         ULONG *idl = txn->mt_free_pgs;
 981                         DPRINTF("IDL write txn %lu root %lu num %lu",
 982                                 txn->mt_txnid, txn->mt_dbs[FREE_DBI].md_root, idl[0]);
 983                         for (i=0; i<idl[0]; i++) {
 984                                 DPRINTF("IDL %lu", idl[i+1]);
 985                         }
 986                 }
 987 #endif
 988                 /* write to last page of freeDB */
 989                 key.mv_size = sizeof(pgno_t);
 990                 key.mv_data = (char *)&txn->mt_txnid;
 991                 data.mv_data = txn->mt_free_pgs;
 992                 /* The free list can still grow during this call,
 993                  * despite the pre-emptive touches above. So check
 994                  * and make sure the entire thing got written.
 995                  */
 996                 do {
 997                         i = txn->mt_free_pgs[0];
 998                         data.mv_size = MDB_IDL_SIZEOF(txn->mt_free_pgs);
 999                         rc = mdb_cursor_put(&mc, &key, &data, 0);
1000                         if (rc) {
1001                                 mdb_txn_abort(txn);
1002                                 return rc;
1003                         }
1004                 } while (i != txn->mt_free_pgs[0]);
1005         }
1006         /* should only be one record now */
1007         if (env->me_pghead) {
1008                 MDB_val key, data;
1009                 MDB_oldpages *mop;
1010
1011                 mop = env->me_pghead;
1012                 key.mv_size = sizeof(pgno_t);
1013                 key.mv_data = (char *)&mop->mo_txnid;
1014                 data.mv_size = MDB_IDL_SIZEOF(mop->mo_pages);
1015                 data.mv_data = mop->mo_pages;
1016                 mdb_cursor_put(&mc, &key, &data, 0);
1017                 free(env->me_pghead);
1018                 env->me_pghead = NULL;
1019         }
1020
1021         /* Update DB root pointers. Their pages have already been
1022          * touched so this is all in-place and cannot fail.
1023          */
1024         {
1025                 MDB_val data;
1026                 data.mv_size = sizeof(MDB_db);
1027
1028                 mc.mc_dbi = MAIN_DBI;
1029                 mc.mc_flags = 0;
1030                 for (i = 2; i < txn->mt_numdbs; i++) {
1031                         if (txn->mt_dbxs[i].md_dirty) {
1032                                 data.mv_data = &txn->mt_dbs[i];
1033                                 mdb_cursor_put(&mc, &txn->mt_dbxs[i].md_name, &data, 0);
1034                         }
1035                 }
1036         }
1037
1038         /* Commit up to MDB_COMMIT_PAGES dirty pages to disk until done.
1039          */
1040         next = 0;
1041         i = 1;
1042         do {
1043 #ifdef _WIN32
1044                 /* Windows actually supports scatter/gather I/O, but only on
1045                  * unbuffered file handles. Since we're relying on the OS page
1046                  * cache for all our data, that's self-defeating. So we just
1047                  * write pages one at a time. We use the ov structure to set
1048                  * the write offset, to at least save the overhead of a Seek
1049                  * system call.
1050                  */
1051                 OVERLAPPED ov;
1052                 memset(&ov, 0, sizeof(ov));
1053                 for (; i<=txn->mt_u.dirty_list[0].mid; i++) {
1054                         size_t wsize;
1055                         dp = txn->mt_u.dirty_list[i].mptr;
1056                         DPRINTF("committing page %lu", dp->mp_pgno);
1057                         size = dp->mp_pgno * env->me_psize;
1058                         ov.Offset = size & 0xffffffff;
1059                         ov.OffsetHigh = size >> 16;
1060                         ov.OffsetHigh >>= 16;
1061                         /* clear dirty flag */
1062                         dp->mp_flags &= ~P_DIRTY;
1063                         wsize = env->me_psize;
1064                         if (IS_OVERFLOW(dp)) wsize *= dp->mp_pages;
1065                         rc = WriteFile(env->me_fd, dp, wsize, NULL, &ov);
1066                         if (!rc) {
1067                                 n = ErrCode();
1068                                 DPRINTF("WriteFile: %d", n);
1069                                 mdb_txn_abort(txn);
1070                                 return n;
1071                         }
1072                 }
1073                 done = 1;;
1074 #else
1075                 struct iovec     iov[MDB_COMMIT_PAGES];
1076                 n = 0;
1077                 done = 1;
1078                 size = 0;
1079                 for (; i<=txn->mt_u.dirty_list[0].mid; i++) {
1080                         dp = txn->mt_u.dirty_list[i].mptr;
1081                         if (dp->mp_pgno != next) {
1082                                 if (n) {
1083                                         DPRINTF("committing %u dirty pages", n);
1084                                         rc = writev(env->me_fd, iov, n);
1085                                         if (rc != size) {
1086                                                 n = ErrCode();
1087                                                 if (rc > 0)
1088                                                         DPUTS("short write, filesystem full?");
1089                                                 else
1090                                                         DPRINTF("writev: %s", strerror(n));
1091                                                 mdb_txn_abort(txn);
1092                                                 return n;
1093                                         }
1094                                         n = 0;
1095                                         size = 0;
1096                                 }
1097                                 lseek(env->me_fd, dp->mp_pgno * env->me_psize, SEEK_SET);
1098                                 next = dp->mp_pgno;
1099                         }
1100                         DPRINTF("committing page %lu", dp->mp_pgno);
1101                         iov[n].iov_len = env->me_psize;
1102                         if (IS_OVERFLOW(dp)) iov[n].iov_len *= dp->mp_pages;
1103                         iov[n].iov_base = dp;
1104                         size += iov[n].iov_len;
1105                         next = dp->mp_pgno + (IS_OVERFLOW(dp) ? dp->mp_pages : 1);
1106                         /* clear dirty flag */
1107                         dp->mp_flags &= ~P_DIRTY;
1108                         if (++n >= MDB_COMMIT_PAGES) {
1109                                 done = 0;
1110                                 i++;
1111                                 break;
1112                         }
1113                 }
1114
1115                 if (n == 0)
1116                         break;
1117
1118                 DPRINTF("committing %u dirty pages", n);
1119                 rc = writev(env->me_fd, iov, n);
1120                 if (rc != size) {
1121                         n = ErrCode();
1122                         if (rc > 0)
1123                                 DPUTS("short write, filesystem full?");
1124                         else
1125                                 DPRINTF("writev: %s", strerror(n));
1126                         mdb_txn_abort(txn);
1127                         return n;
1128                 }
1129 #endif
1130         } while (!done);
1131
1132         /* Drop the dirty pages.
1133          */
1134         for (i=1; i<=txn->mt_u.dirty_list[0].mid; i++) {
1135                 dp = txn->mt_u.dirty_list[i].mptr;
1136                 if (!IS_OVERFLOW(dp) || dp->mp_pages == 1) {
1137                         dp->mp_next = txn->mt_env->me_dpages;
1138                         txn->mt_env->me_dpages = dp;
1139                 } else {
1140                         free(dp);
1141                 }
1142                 txn->mt_u.dirty_list[i].mid = 0;
1143         }
1144         txn->mt_u.dirty_list[0].mid = 0;
1145
1146         if ((n = mdb_env_sync(env, 0)) != 0 ||
1147             (n = mdb_env_write_meta(txn)) != MDB_SUCCESS) {
1148                 mdb_txn_abort(txn);
1149                 return n;
1150         }
1151
1152 done:
1153         env->me_txn = NULL;
1154         /* update the DB tables */
1155         {
1156                 int toggle = !env->me_db_toggle;
1157                 MDB_db *ip, *jp;
1158
1159                 ip = &env->me_dbs[toggle][2];
1160                 jp = &txn->mt_dbs[2];
1161                 LAZY_RWLOCK_WRLOCK(&env->me_dblock);
1162                 for (i = 2; i < txn->mt_numdbs; i++) {
1163                         if (ip->md_root != jp->md_root)
1164                                 *ip = *jp;
1165                         ip++; jp++;
1166                 }
1167
1168                 for (i = 2; i < txn->mt_numdbs; i++) {
1169                         if (txn->mt_dbxs[i].md_dirty)
1170                                 txn->mt_dbxs[i].md_dirty = 0;
1171                 }
1172                 env->me_db_toggle = toggle;
1173                 env->me_numdbs = txn->mt_numdbs;
1174                 LAZY_RWLOCK_UNLOCK(&env->me_dblock);
1175         }
1176
1177         UNLOCK_MUTEX_W(env);
1178         free(txn);
1179
1180         return MDB_SUCCESS;
1181 }
1182
1183 static int
1184 mdb_env_read_header(MDB_env *env, MDB_meta *meta)
1185 {
1186         char             page[PAGESIZE];
1187         MDB_page        *p;
1188         MDB_meta        *m;
1189         int              rc, err;
1190
1191         /* We don't know the page size yet, so use a minimum value.
1192          */
1193
1194 #ifdef _WIN32
1195         if (!ReadFile(env->me_fd, page, PAGESIZE, (DWORD *)&rc, NULL) || rc == 0)
1196 #else
1197         if ((rc = read(env->me_fd, page, PAGESIZE)) == 0)
1198 #endif
1199         {
1200                 return ENOENT;
1201         }
1202         else if (rc != PAGESIZE) {
1203                 err = ErrCode();
1204                 if (rc > 0)
1205                         err = EINVAL;
1206                 DPRINTF("read: %s", strerror(err));
1207                 return err;
1208         }
1209
1210         p = (MDB_page *)page;
1211
1212         if (!F_ISSET(p->mp_flags, P_META)) {
1213                 DPRINTF("page %lu not a meta page", p->mp_pgno);
1214                 return EINVAL;
1215         }
1216
1217         m = METADATA(p);
1218         if (m->mm_magic != MDB_MAGIC) {
1219                 DPUTS("meta has invalid magic");
1220                 return EINVAL;
1221         }
1222
1223         if (m->mm_version != MDB_VERSION) {
1224                 DPRINTF("database is version %u, expected version %u",
1225                     m->mm_version, MDB_VERSION);
1226                 return MDB_VERSION_MISMATCH;
1227         }
1228
1229         memcpy(meta, m, sizeof(*m));
1230         return 0;
1231 }
1232
1233 static int
1234 mdb_env_init_meta(MDB_env *env, MDB_meta *meta)
1235 {
1236         MDB_page *p, *q;
1237         MDB_meta *m;
1238         int rc;
1239         unsigned int     psize;
1240
1241         DPUTS("writing new meta page");
1242
1243         GetPageSize(psize);
1244
1245         meta->mm_magic = MDB_MAGIC;
1246         meta->mm_version = MDB_VERSION;
1247         meta->mm_psize = psize;
1248         meta->mm_last_pg = 1;
1249         meta->mm_flags = env->me_flags & 0xffff;
1250         meta->mm_flags |= MDB_INTEGERKEY;
1251         meta->mm_dbs[0].md_root = P_INVALID;
1252         meta->mm_dbs[1].md_root = P_INVALID;
1253
1254         p = calloc(2, psize);
1255         p->mp_pgno = 0;
1256         p->mp_flags = P_META;
1257
1258         m = METADATA(p);
1259         memcpy(m, meta, sizeof(*meta));
1260
1261         q = (MDB_page *)((char *)p + psize);
1262
1263         q->mp_pgno = 1;
1264         q->mp_flags = P_META;
1265
1266         m = METADATA(q);
1267         memcpy(m, meta, sizeof(*meta));
1268
1269 #ifdef _WIN32
1270         {
1271                 DWORD len;
1272                 rc = WriteFile(env->me_fd, p, psize * 2, &len, NULL);
1273                 rc = (len == psize * 2) ? MDB_SUCCESS : ErrCode();
1274         }
1275 #else
1276         rc = write(env->me_fd, p, psize * 2);
1277         rc = (rc == (int)psize * 2) ? MDB_SUCCESS : ErrCode();
1278 #endif
1279         free(p);
1280         return rc;
1281 }
1282
1283 static int
1284 mdb_env_write_meta(MDB_txn *txn)
1285 {
1286         MDB_env *env;
1287         MDB_meta        meta, metab;
1288         off_t off;
1289         int rc, len, toggle;
1290         char *ptr;
1291 #ifdef _WIN32
1292         OVERLAPPED ov;
1293 #endif
1294
1295         assert(txn != NULL);
1296         assert(txn->mt_env != NULL);
1297
1298         toggle = !txn->mt_toggle;
1299         DPRINTF("writing meta page %d for root page %lu",
1300                 toggle, txn->mt_dbs[MAIN_DBI].md_root);
1301
1302         env = txn->mt_env;
1303
1304         metab.mm_txnid = env->me_metas[toggle]->mm_txnid;
1305         metab.mm_last_pg = env->me_metas[toggle]->mm_last_pg;
1306
1307         ptr = (char *)&meta;
1308         off = offsetof(MDB_meta, mm_dbs[0].md_depth);
1309         len = sizeof(MDB_meta) - off;
1310
1311         ptr += off;
1312         meta.mm_dbs[0] = txn->mt_dbs[0];
1313         meta.mm_dbs[1] = txn->mt_dbs[1];
1314         meta.mm_last_pg = txn->mt_next_pgno - 1;
1315         meta.mm_txnid = txn->mt_txnid;
1316
1317         if (toggle)
1318                 off += env->me_psize;
1319         off += PAGEHDRSZ;
1320
1321         /* Write to the SYNC fd */
1322 #ifdef _WIN32
1323         {
1324                 memset(&ov, 0, sizeof(ov));
1325                 ov.Offset = off;
1326                 WriteFile(env->me_mfd, ptr, len, (DWORD *)&rc, &ov);
1327         }
1328 #else
1329         rc = pwrite(env->me_mfd, ptr, len, off);
1330 #endif
1331         if (rc != len) {
1332                 int r2;
1333                 rc = ErrCode();
1334                 DPUTS("write failed, disk error?");
1335                 /* On a failure, the pagecache still contains the new data.
1336                  * Write some old data back, to prevent it from being used.
1337                  * Use the non-SYNC fd; we know it will fail anyway.
1338                  */
1339                 meta.mm_last_pg = metab.mm_last_pg;
1340                 meta.mm_txnid = metab.mm_txnid;
1341 #ifdef _WIN32
1342                 WriteFile(env->me_fd, ptr, len, NULL, &ov);
1343 #else
1344                 r2 = pwrite(env->me_fd, ptr, len, off);
1345 #endif
1346                 env->me_flags |= MDB_FATAL_ERROR;
1347                 return rc;
1348         }
1349         /* Memory ordering issues are irrelevant; since the entire writer
1350          * is wrapped by wmutex, all of these changes will become visible
1351          * after the wmutex is unlocked. Since the DB is multi-version,
1352          * readers will get consistent data regardless of how fresh or
1353          * how stale their view of these values is.
1354          */
1355         LAZY_MUTEX_LOCK(&env->me_txns->mti_mutex);
1356         txn->mt_env->me_txns->mti_me_toggle = toggle;
1357         txn->mt_env->me_txns->mti_txnid = txn->mt_txnid;
1358         LAZY_MUTEX_UNLOCK(&env->me_txns->mti_mutex);
1359
1360         return MDB_SUCCESS;
1361 }
1362
1363 static int
1364 mdb_env_read_meta(MDB_env *env, int *which)
1365 {
1366         int toggle = 0;
1367
1368         assert(env != NULL);
1369
1370         if (env->me_metas[0]->mm_txnid < env->me_metas[1]->mm_txnid)
1371                 toggle = 1;
1372
1373         DPRINTF("Using meta page %d", toggle);
1374         *which = toggle;
1375
1376         return MDB_SUCCESS;
1377 }
1378
1379 int
1380 mdb_env_create(MDB_env **env)
1381 {
1382         MDB_env *e;
1383
1384         e = calloc(1, sizeof(MDB_env));
1385         if (!e) return ENOMEM;
1386
1387         e->me_maxreaders = DEFAULT_READERS;
1388         e->me_maxdbs = 2;
1389         e->me_fd = INVALID_HANDLE_VALUE;
1390         e->me_lfd = INVALID_HANDLE_VALUE;
1391         e->me_mfd = INVALID_HANDLE_VALUE;
1392         *env = e;
1393         return MDB_SUCCESS;
1394 }
1395
1396 int
1397 mdb_env_set_mapsize(MDB_env *env, size_t size)
1398 {
1399         if (env->me_map)
1400                 return EINVAL;
1401         env->me_mapsize = size;
1402         return MDB_SUCCESS;
1403 }
1404
1405 int
1406 mdb_env_set_maxdbs(MDB_env *env, int dbs)
1407 {
1408         if (env->me_map)
1409                 return EINVAL;
1410         env->me_maxdbs = dbs;
1411         return MDB_SUCCESS;
1412 }
1413
1414 int
1415 mdb_env_set_maxreaders(MDB_env *env, int readers)
1416 {
1417         if (env->me_map)
1418                 return EINVAL;
1419         env->me_maxreaders = readers;
1420         return MDB_SUCCESS;
1421 }
1422
1423 int
1424 mdb_env_get_maxreaders(MDB_env *env, int *readers)
1425 {
1426         if (!env || !readers)
1427                 return EINVAL;
1428         *readers = env->me_maxreaders;
1429         return MDB_SUCCESS;
1430 }
1431
1432 static int
1433 mdb_env_open2(MDB_env *env, unsigned int flags)
1434 {
1435         int i, newenv = 0, toggle;
1436         MDB_meta meta;
1437         MDB_page *p;
1438
1439         env->me_flags = flags;
1440
1441         memset(&meta, 0, sizeof(meta));
1442
1443         if ((i = mdb_env_read_header(env, &meta)) != 0) {
1444                 if (i != ENOENT)
1445                         return i;
1446                 DPUTS("new mdbenv");
1447                 newenv = 1;
1448         }
1449
1450         if (!env->me_mapsize) {
1451                 env->me_mapsize = newenv ? DEFAULT_MAPSIZE : meta.mm_mapsize;
1452         }
1453
1454 #ifdef _WIN32
1455         {
1456                 HANDLE mh;
1457                 LONG sizelo, sizehi;
1458                 sizelo = env->me_mapsize & 0xffffffff;
1459                 sizehi = env->me_mapsize >> 16;         /* pointless on WIN32, only needed on W64 */
1460                 sizehi >>= 16;
1461                 /* Windows won't create mappings for zero length files.
1462                  * Just allocate the maxsize right now.
1463                  */
1464                 if (newenv) {
1465                         SetFilePointer(env->me_fd, sizelo, sizehi ? &sizehi : NULL, 0);
1466                         if (!SetEndOfFile(env->me_fd))
1467                                 return ErrCode();
1468                         SetFilePointer(env->me_fd, 0, NULL, 0);
1469                 }
1470                 mh = CreateFileMapping(env->me_fd, NULL, PAGE_READONLY,
1471                         sizehi, sizelo, NULL);
1472                 if (!mh)
1473                         return ErrCode();
1474                 env->me_map = MapViewOfFileEx(mh, FILE_MAP_READ, 0, 0, env->me_mapsize,
1475                         meta.mm_address);
1476                 CloseHandle(mh);
1477                 if (!env->me_map)
1478                         return ErrCode();
1479         }
1480 #else
1481         i = MAP_SHARED;
1482         if (meta.mm_address && (flags & MDB_FIXEDMAP))
1483                 i |= MAP_FIXED;
1484         env->me_map = mmap(meta.mm_address, env->me_mapsize, PROT_READ, i,
1485                 env->me_fd, 0);
1486         if (env->me_map == MAP_FAILED)
1487                 return ErrCode();
1488 #endif
1489
1490         if (newenv) {
1491                 meta.mm_mapsize = env->me_mapsize;
1492                 if (flags & MDB_FIXEDMAP)
1493                         meta.mm_address = env->me_map;
1494                 i = mdb_env_init_meta(env, &meta);
1495                 if (i != MDB_SUCCESS) {
1496                         munmap(env->me_map, env->me_mapsize);
1497                         return i;
1498                 }
1499         }
1500         env->me_psize = meta.mm_psize;
1501
1502         env->me_maxpg = env->me_mapsize / env->me_psize;
1503
1504         p = (MDB_page *)env->me_map;
1505         env->me_metas[0] = METADATA(p);
1506         env->me_metas[1] = (MDB_meta *)((char *)env->me_metas[0] + meta.mm_psize);
1507
1508         if ((i = mdb_env_read_meta(env, &toggle)) != 0)
1509                 return i;
1510
1511         DPRINTF("opened database version %u, pagesize %u",
1512             env->me_metas[toggle]->mm_version, env->me_psize);
1513         DPRINTF("depth: %u", env->me_metas[toggle]->mm_dbs[MAIN_DBI].md_depth);
1514         DPRINTF("entries: %lu", env->me_metas[toggle]->mm_dbs[MAIN_DBI].md_entries);
1515         DPRINTF("branch pages: %lu", env->me_metas[toggle]->mm_dbs[MAIN_DBI].md_branch_pages);
1516         DPRINTF("leaf pages: %lu", env->me_metas[toggle]->mm_dbs[MAIN_DBI].md_leaf_pages);
1517         DPRINTF("overflow pages: %lu", env->me_metas[toggle]->mm_dbs[MAIN_DBI].md_overflow_pages);
1518         DPRINTF("root: %lu", env->me_metas[toggle]->mm_dbs[MAIN_DBI].md_root);
1519
1520         return MDB_SUCCESS;
1521 }
1522
1523 #ifndef _WIN32
1524 /* Windows doesn't support destructor callbacks for thread-specific storage */
1525 static void
1526 mdb_env_reader_dest(void *ptr)
1527 {
1528         MDB_reader *reader = ptr;
1529
1530         reader->mr_txnid = 0;
1531         reader->mr_pid = 0;
1532         reader->mr_tid = 0;
1533 }
1534 #endif
1535
1536 /* downgrade the exclusive lock on the region back to shared */
1537 static void
1538 mdb_env_share_locks(MDB_env *env)
1539 {
1540         int toggle = 0;
1541
1542         if (env->me_metas[0]->mm_txnid < env->me_metas[1]->mm_txnid)
1543                 toggle = 1;
1544         env->me_txns->mti_me_toggle = toggle;
1545         env->me_txns->mti_txnid = env->me_metas[toggle]->mm_txnid;
1546
1547 #ifdef _WIN32
1548         {
1549                 OVERLAPPED ov;
1550                 /* First acquire a shared lock. The Unlock will
1551                  * then release the existing exclusive lock.
1552                  */
1553                 memset(&ov, 0, sizeof(ov));
1554                 LockFileEx(env->me_lfd, 0, 0, 1, 0, &ov);
1555                 UnlockFile(env->me_lfd, 0, 0, 1, 0);
1556         }
1557 #else
1558         {
1559                 struct flock lock_info;
1560                 /* The shared lock replaces the existing lock */
1561                 memset((void *)&lock_info, 0, sizeof(lock_info));
1562                 lock_info.l_type = F_RDLCK;
1563                 lock_info.l_whence = SEEK_SET;
1564                 lock_info.l_start = 0;
1565                 lock_info.l_len = 1;
1566                 fcntl(env->me_lfd, F_SETLK, &lock_info);
1567         }
1568 #endif
1569 }
1570
1571 static int
1572 mdb_env_setup_locks(MDB_env *env, char *lpath, int mode, int *excl)
1573 {
1574         int rc;
1575         off_t size, rsize;
1576
1577         *excl = 0;
1578
1579 #ifdef _WIN32
1580         if ((env->me_lfd = CreateFile(lpath, GENERIC_READ|GENERIC_WRITE,
1581                 FILE_SHARE_READ|FILE_SHARE_WRITE, NULL, OPEN_ALWAYS,
1582                 FILE_ATTRIBUTE_NORMAL, NULL)) == INVALID_HANDLE_VALUE) {
1583                 rc = ErrCode();
1584                 return rc;
1585         }
1586         /* Try to get exclusive lock. If we succeed, then
1587          * nobody is using the lock region and we should initialize it.
1588          */
1589         {
1590                 if (LockFile(env->me_lfd, 0, 0, 1, 0)) {
1591                         *excl = 1;
1592                 } else {
1593                         OVERLAPPED ov;
1594                         memset(&ov, 0, sizeof(ov));
1595                         if (!LockFileEx(env->me_lfd, 0, 0, 1, 0, &ov)) {
1596                                 rc = ErrCode();
1597                                 goto fail;
1598                         }
1599                 }
1600         }
1601         size = GetFileSize(env->me_lfd, NULL);
1602 #else
1603         if ((env->me_lfd = open(lpath, O_RDWR|O_CREAT, mode)) == -1) {
1604                 rc = ErrCode();
1605                 return rc;
1606         }
1607         /* Try to get exclusive lock. If we succeed, then
1608          * nobody is using the lock region and we should initialize it.
1609          */
1610         {
1611                 struct flock lock_info;
1612                 memset((void *)&lock_info, 0, sizeof(lock_info));
1613                 lock_info.l_type = F_WRLCK;
1614                 lock_info.l_whence = SEEK_SET;
1615                 lock_info.l_start = 0;
1616                 lock_info.l_len = 1;
1617                 rc = fcntl(env->me_lfd, F_SETLK, &lock_info);
1618                 if (rc == 0) {
1619                         *excl = 1;
1620                 } else {
1621                         lock_info.l_type = F_RDLCK;
1622                         rc = fcntl(env->me_lfd, F_SETLKW, &lock_info);
1623                         if (rc) {
1624                                 rc = ErrCode();
1625                                 goto fail;
1626                         }
1627                 }
1628         }
1629         size = lseek(env->me_lfd, 0, SEEK_END);
1630 #endif
1631         rsize = (env->me_maxreaders-1) * sizeof(MDB_reader) + sizeof(MDB_txninfo);
1632         if (size < rsize && *excl) {
1633 #ifdef _WIN32
1634                 SetFilePointer(env->me_lfd, rsize, NULL, 0);
1635                 if (!SetEndOfFile(env->me_lfd)) {
1636                         rc = ErrCode();
1637                         goto fail;
1638                 }
1639 #else
1640                 if (ftruncate(env->me_lfd, rsize) != 0) {
1641                         rc = ErrCode();
1642                         goto fail;
1643                 }
1644 #endif
1645         } else {
1646                 rsize = size;
1647                 size = rsize - sizeof(MDB_txninfo);
1648                 env->me_maxreaders = size/sizeof(MDB_reader) + 1;
1649         }
1650 #ifdef _WIN32
1651         {
1652                 HANDLE mh;
1653                 mh = CreateFileMapping(env->me_lfd, NULL, PAGE_READWRITE,
1654                         0, 0, NULL);
1655                 if (!mh) {
1656                         rc = ErrCode();
1657                         goto fail;
1658                 }
1659                 env->me_txns = MapViewOfFileEx(mh, FILE_MAP_WRITE, 0, 0, rsize, NULL);
1660                 CloseHandle(mh);
1661                 if (!env->me_txns) {
1662                         rc = ErrCode();
1663                         goto fail;
1664                 }
1665         }
1666 #else
1667         env->me_txns = mmap(0, rsize, PROT_READ|PROT_WRITE, MAP_SHARED,
1668                 env->me_lfd, 0);
1669         if (env->me_txns == MAP_FAILED) {
1670                 rc = ErrCode();
1671                 goto fail;
1672         }
1673 #endif
1674         if (*excl) {
1675 #ifdef _WIN32
1676                 char *ptr;
1677                 if (!mdb_sec_inited) {
1678                         InitializeSecurityDescriptor(&mdb_null_sd,
1679                                 SECURITY_DESCRIPTOR_REVISION);
1680                         SetSecurityDescriptorDacl(&mdb_null_sd, TRUE, 0, FALSE);
1681                         mdb_all_sa.nLength = sizeof(SECURITY_ATTRIBUTES);
1682                         mdb_all_sa.bInheritHandle = FALSE;
1683                         mdb_all_sa.lpSecurityDescriptor = &mdb_null_sd;
1684                         mdb_sec_inited = 1;
1685                 }
1686                 /* FIXME: only using up to 20 characters of the env path here,
1687                  * probably not enough to assure uniqueness...
1688                  */
1689                 sprintf(env->me_txns->mti_rmname, "Global\\MDBr%.20s", lpath);
1690                 ptr = env->me_txns->mti_rmname + sizeof("Global\\MDBr");
1691                 while ((ptr = strchr(ptr, '\\')))
1692                         *ptr++ = '/';
1693                 env->me_rmutex = CreateMutex(&mdb_all_sa, FALSE, env->me_txns->mti_rmname);
1694                 if (!env->me_rmutex) {
1695                         rc = ErrCode();
1696                         goto fail;
1697                 }
1698                 sprintf(env->me_txns->mti_rmname, "Global\\MDBw%.20s", lpath);
1699                 ptr = env->me_txns->mti_rmname + sizeof("Global\\MDBw");
1700                 while ((ptr = strchr(ptr, '\\')))
1701                         *ptr++ = '/';
1702                 env->me_wmutex = CreateMutex(&mdb_all_sa, FALSE, env->me_txns->mti_rmname);
1703                 if (!env->me_wmutex) {
1704                         rc = ErrCode();
1705                         goto fail;
1706                 }
1707 #else
1708                 pthread_mutexattr_t mattr;
1709
1710                 pthread_mutexattr_init(&mattr);
1711                 rc = pthread_mutexattr_setpshared(&mattr, PTHREAD_PROCESS_SHARED);
1712                 if (rc) {
1713                         goto fail;
1714                 }
1715                 pthread_mutex_init(&env->me_txns->mti_mutex, &mattr);
1716                 pthread_mutex_init(&env->me_txns->mti_wmutex, &mattr);
1717 #endif
1718                 env->me_txns->mti_version = MDB_VERSION;
1719                 env->me_txns->mti_magic = MDB_MAGIC;
1720                 env->me_txns->mti_txnid = 0;
1721                 env->me_txns->mti_numreaders = 0;
1722                 env->me_txns->mti_me_toggle = 0;
1723
1724         } else {
1725                 if (env->me_txns->mti_magic != MDB_MAGIC) {
1726                         DPUTS("lock region has invalid magic");
1727                         rc = EINVAL;
1728                         goto fail;
1729                 }
1730                 if (env->me_txns->mti_version != MDB_VERSION) {
1731                         DPRINTF("lock region is version %u, expected version %u",
1732                                 env->me_txns->mti_version, MDB_VERSION);
1733                         rc = MDB_VERSION_MISMATCH;
1734                         goto fail;
1735                 }
1736                 rc = ErrCode();
1737                 if (rc != EACCES && rc != EAGAIN) {
1738                         goto fail;
1739                 }
1740 #ifdef _WIN32
1741                 env->me_rmutex = OpenMutex(SYNCHRONIZE, FALSE, env->me_txns->mti_rmname);
1742                 if (!env->me_rmutex) {
1743                         rc = ErrCode();
1744                         goto fail;
1745                 }
1746                 env->me_wmutex = OpenMutex(SYNCHRONIZE, FALSE, env->me_txns->mti_wmname);
1747                 if (!env->me_wmutex) {
1748                         rc = ErrCode();
1749                         goto fail;
1750                 }
1751 #endif
1752         }
1753         return MDB_SUCCESS;
1754
1755 fail:
1756         close(env->me_lfd);
1757         env->me_lfd = INVALID_HANDLE_VALUE;
1758         return rc;
1759
1760 }
1761
1762 #define LOCKNAME        "/lock.mdb"
1763 #define DATANAME        "/data.mdb"
1764 int
1765 mdb_env_open(MDB_env *env, const char *path, unsigned int flags, mode_t mode)
1766 {
1767         int             oflags, rc, len, excl;
1768         char *lpath, *dpath;
1769
1770         len = strlen(path);
1771         lpath = malloc(len + sizeof(LOCKNAME) + len + sizeof(DATANAME));
1772         if (!lpath)
1773                 return ENOMEM;
1774         dpath = lpath + len + sizeof(LOCKNAME);
1775         sprintf(lpath, "%s" LOCKNAME, path);
1776         sprintf(dpath, "%s" DATANAME, path);
1777
1778         rc = mdb_env_setup_locks(env, lpath, mode, &excl);
1779         if (rc)
1780                 goto leave;
1781
1782 #ifdef _WIN32
1783         if (F_ISSET(flags, MDB_RDONLY)) {
1784                 oflags = GENERIC_READ;
1785                 len = OPEN_EXISTING;
1786         } else {
1787                 oflags = GENERIC_READ|GENERIC_WRITE;
1788                 len = OPEN_ALWAYS;
1789         }
1790         mode = FILE_ATTRIBUTE_NORMAL;
1791         if ((env->me_fd = CreateFile(dpath, oflags, FILE_SHARE_READ|FILE_SHARE_WRITE,
1792                         NULL, len, mode, NULL)) == INVALID_HANDLE_VALUE) {
1793                 rc = ErrCode();
1794                 goto leave;
1795         }
1796 #else
1797         if (F_ISSET(flags, MDB_RDONLY))
1798                 oflags = O_RDONLY;
1799         else
1800                 oflags = O_RDWR | O_CREAT;
1801
1802         if ((env->me_fd = open(dpath, oflags, mode)) == -1) {
1803                 rc = ErrCode();
1804                 goto leave;
1805         }
1806 #endif
1807
1808         if ((rc = mdb_env_open2(env, flags)) == MDB_SUCCESS) {
1809                 /* synchronous fd for meta writes */
1810 #ifdef _WIN32
1811                 if (!(flags & (MDB_RDONLY|MDB_NOSYNC)))
1812                         mode |= FILE_FLAG_WRITE_THROUGH;
1813                 if ((env->me_mfd = CreateFile(dpath, oflags, FILE_SHARE_READ|FILE_SHARE_WRITE,
1814                         NULL, len, mode, NULL)) == INVALID_HANDLE_VALUE) {
1815                         rc = ErrCode();
1816                         goto leave;
1817                 }
1818 #else
1819                 if (!(flags & (MDB_RDONLY|MDB_NOSYNC)))
1820                         oflags |= MDB_DSYNC;
1821                 if ((env->me_mfd = open(dpath, oflags, mode)) == -1) {
1822                         rc = ErrCode();
1823                         goto leave;
1824                 }
1825 #endif
1826                 env->me_path = strdup(path);
1827                 DPRINTF("opened dbenv %p", (void *) env);
1828                 pthread_key_create(&env->me_txkey, mdb_env_reader_dest);
1829                 LAZY_RWLOCK_INIT(&env->me_dblock, NULL);
1830                 if (excl)
1831                         mdb_env_share_locks(env);
1832                 env->me_dbxs = calloc(env->me_maxdbs, sizeof(MDB_dbx));
1833                 env->me_dbs[0] = calloc(env->me_maxdbs, sizeof(MDB_db));
1834                 env->me_dbs[1] = calloc(env->me_maxdbs, sizeof(MDB_db));
1835                 env->me_numdbs = 2;
1836         }
1837
1838 leave:
1839         if (rc) {
1840                 if (env->me_fd != INVALID_HANDLE_VALUE) {
1841                         close(env->me_fd);
1842                         env->me_fd = INVALID_HANDLE_VALUE;
1843                 }
1844                 if (env->me_lfd != INVALID_HANDLE_VALUE) {
1845                         close(env->me_lfd);
1846                         env->me_lfd = INVALID_HANDLE_VALUE;
1847                 }
1848         }
1849         free(lpath);
1850         return rc;
1851 }
1852
1853 void
1854 mdb_env_close(MDB_env *env)
1855 {
1856         MDB_page *dp;
1857
1858         if (env == NULL)
1859                 return;
1860
1861         while (env->me_dpages) {
1862                 dp = env->me_dpages;
1863                 env->me_dpages = dp->mp_next;
1864                 free(dp);
1865         }
1866
1867         free(env->me_dbs[1]);
1868         free(env->me_dbs[0]);
1869         free(env->me_dbxs);
1870         free(env->me_path);
1871
1872         LAZY_RWLOCK_DESTROY(&env->me_dblock);
1873         pthread_key_delete(env->me_txkey);
1874
1875         if (env->me_map) {
1876                 munmap(env->me_map, env->me_mapsize);
1877         }
1878         close(env->me_mfd);
1879         close(env->me_fd);
1880         if (env->me_txns) {
1881                 pid_t pid = getpid();
1882                 unsigned int i;
1883                 for (i=0; i<env->me_txns->mti_numreaders; i++)
1884                         if (env->me_txns->mti_readers[i].mr_pid == pid)
1885                                 env->me_txns->mti_readers[i].mr_pid = 0;
1886                 munmap(env->me_txns, (env->me_maxreaders-1)*sizeof(MDB_reader)+sizeof(MDB_txninfo));
1887         }
1888         close(env->me_lfd);
1889         free(env);
1890 }
1891
1892 /* only for aligned ints */
1893 static int
1894 intcmp(const MDB_val *a, const MDB_val *b)
1895 {
1896         if (a->mv_size == sizeof(long))
1897         {
1898                 unsigned long *la, *lb;
1899                 la = a->mv_data;
1900                 lb = b->mv_data;
1901                 return *la - *lb;
1902         } else {
1903                 unsigned int *ia, *ib;
1904                 ia = a->mv_data;
1905                 ib = b->mv_data;
1906                 return *ia - *ib;
1907         }
1908 }
1909
1910 /* ints must always be the same size */
1911 static int
1912 cintcmp(const MDB_val *a, const MDB_val *b)
1913 {
1914 #if __BYTE_ORDER == __LITTLE_ENDIAN
1915         unsigned char *u, *c;
1916         int x;
1917
1918         u = a->mv_data + a->mv_size;
1919         c = b->mv_data + a->mv_size;
1920         do {
1921                 x = *--u - *--c;
1922         } while(!x && u > (unsigned char *)a->mv_data);
1923         return x;
1924 #else
1925         return memcmp(a->mv_data, b->mv_data, a->mv_size);
1926 #endif
1927 }
1928
1929 static int
1930 memncmp(const MDB_val *a, const MDB_val *b)
1931 {
1932         int diff, len_diff;
1933         unsigned int len;
1934
1935         len = a->mv_size;
1936         len_diff = a->mv_size - b->mv_size;
1937         if (len_diff > 0)
1938                 len = b->mv_size;
1939         diff = memcmp(a->mv_data, b->mv_data, len);
1940         return diff ? diff : len_diff;
1941 }
1942
1943 static int
1944 memnrcmp(const MDB_val *a, const MDB_val *b)
1945 {
1946         const unsigned char     *p1, *p2, *p1_lim;
1947         int diff, len_diff;
1948
1949         if (b->mv_size == 0)
1950                 return a->mv_size != 0;
1951         if (a->mv_size == 0)
1952                 return -1;
1953
1954         p1 = (const unsigned char *)a->mv_data + a->mv_size - 1;
1955         p2 = (const unsigned char *)b->mv_data + b->mv_size - 1;
1956
1957         len_diff = a->mv_size - b->mv_size;
1958         if (len_diff < 0)
1959                 p1_lim = p1 - a->mv_size;
1960         else
1961                 p1_lim = p1 - b->mv_size;
1962
1963         while (p1 > p1_lim) {
1964                 diff = *p1 - *p2;
1965                 if (diff)
1966                         return diff;
1967                 p1--;
1968                 p2--;
1969         }
1970         return len_diff;
1971 }
1972
1973 /* Search for key within a leaf page, using binary search.
1974  * Returns the smallest entry larger or equal to the key.
1975  * If exactp is non-null, stores whether the found entry was an exact match
1976  * in *exactp (1 or 0).
1977  * If kip is non-null, stores the index of the found entry in *kip.
1978  * If no entry larger or equal to the key is found, returns NULL.
1979  */
1980 static MDB_node *
1981 mdb_search_node(MDB_cursor *mc, MDB_val *key, int *exactp)
1982 {
1983         unsigned int     i = 0, nkeys;
1984         int              low, high;
1985         int              rc = 0;
1986         MDB_page *mp = mc->mc_pg[mc->mc_top];
1987         MDB_node        *node = NULL;
1988         MDB_val  nodekey;
1989         MDB_cmp_func *cmp;
1990         DKBUF;
1991
1992         nkeys = NUMKEYS(mp);
1993
1994         DPRINTF("searching %u keys in %s page %lu",
1995             nkeys, IS_LEAF(mp) ? "leaf" : "branch",
1996             mp->mp_pgno);
1997
1998         assert(nkeys > 0);
1999
2000         low = IS_LEAF(mp) ? 0 : 1;
2001         high = nkeys - 1;
2002         cmp = mc->mc_txn->mt_dbxs[mc->mc_dbi].md_cmp;
2003         if (IS_LEAF2(mp)) {
2004                 nodekey.mv_size = mc->mc_txn->mt_dbs[mc->mc_dbi].md_pad;
2005                 node = NODEPTR(mp, 0);  /* fake */
2006         }
2007         while (low <= high) {
2008                 i = (low + high) >> 1;
2009
2010                 if (IS_LEAF2(mp)) {
2011                         nodekey.mv_data = LEAF2KEY(mp, i, nodekey.mv_size);
2012                 } else {
2013                         node = NODEPTR(mp, i);
2014
2015                         nodekey.mv_size = node->mn_ksize;
2016                         nodekey.mv_data = NODEKEY(node);
2017                 }
2018
2019                 rc = cmp(key, &nodekey);
2020
2021 #if DEBUG
2022                 if (IS_LEAF(mp))
2023                         DPRINTF("found leaf index %u [%s], rc = %i",
2024                             i, DKEY(&nodekey), rc);
2025                 else
2026                         DPRINTF("found branch index %u [%s -> %lu], rc = %i",
2027                             i, DKEY(&nodekey), NODEPGNO(node), rc);
2028 #endif
2029
2030                 if (rc == 0)
2031                         break;
2032                 if (rc > 0)
2033                         low = i + 1;
2034                 else
2035                         high = i - 1;
2036         }
2037
2038         if (rc > 0) {   /* Found entry is less than the key. */
2039                 i++;    /* Skip to get the smallest entry larger than key. */
2040                 if (!IS_LEAF2(mp))
2041                         node = NODEPTR(mp, i);
2042         }
2043         if (exactp)
2044                 *exactp = (rc == 0);
2045         /* store the key index */
2046         mc->mc_ki[mc->mc_top] = i;
2047         if (i >= nkeys)
2048                 /* There is no entry larger or equal to the key. */
2049                 return NULL;
2050
2051         /* nodeptr is fake for LEAF2 */
2052         return node;
2053 }
2054
2055 static void
2056 cursor_pop_page(MDB_cursor *mc)
2057 {
2058         MDB_page        *top;
2059
2060         if (mc->mc_snum) {
2061                 top = mc->mc_pg[mc->mc_top];
2062                 mc->mc_snum--;
2063                 if (mc->mc_snum)
2064                         mc->mc_top--;
2065
2066                 DPRINTF("popped page %lu off db %u cursor %p", top->mp_pgno,
2067                         mc->mc_dbi, (void *) mc);
2068         }
2069 }
2070
2071 static int
2072 cursor_push_page(MDB_cursor *mc, MDB_page *mp)
2073 {
2074         DPRINTF("pushing page %lu on db %u cursor %p", mp->mp_pgno,
2075                 mc->mc_dbi, (void *) mc);
2076
2077         if (mc->mc_snum >= CURSOR_STACK)
2078                 return ENOMEM;
2079
2080         mc->mc_top = mc->mc_snum++;
2081         mc->mc_pg[mc->mc_top] = mp;
2082         mc->mc_ki[mc->mc_top] = 0;
2083
2084         return MDB_SUCCESS;
2085 }
2086
2087 static int
2088 mdb_get_page(MDB_txn *txn, pgno_t pgno, MDB_page **ret)
2089 {
2090         MDB_page *p = NULL;
2091
2092         if (!F_ISSET(txn->mt_flags, MDB_TXN_RDONLY) && txn->mt_u.dirty_list[0].mid) {
2093                 unsigned x;
2094                 x = mdb_mid2l_search(txn->mt_u.dirty_list, pgno);
2095                 if (x <= txn->mt_u.dirty_list[0].mid && txn->mt_u.dirty_list[x].mid == pgno) {
2096                         p = txn->mt_u.dirty_list[x].mptr;
2097                 }
2098         }
2099         if (!p) {
2100                 if (pgno <= txn->mt_env->me_metas[txn->mt_toggle]->mm_last_pg)
2101                         p = (MDB_page *)(txn->mt_env->me_map + txn->mt_env->me_psize * pgno);
2102         }
2103         *ret = p;
2104         if (!p) {
2105                 DPRINTF("page %lu not found", pgno);
2106                 assert(p != NULL);
2107         }
2108         return (p != NULL) ? MDB_SUCCESS : MDB_PAGE_NOTFOUND;
2109 }
2110
2111 static int
2112 mdb_search_page_root(MDB_cursor *mc, MDB_val *key, int modify)
2113 {
2114         MDB_page        *mp = mc->mc_pg[mc->mc_top];
2115         DKBUF;
2116         int rc;
2117
2118
2119         while (IS_BRANCH(mp)) {
2120                 MDB_node        *node;
2121
2122                 DPRINTF("branch page %lu has %u keys", mp->mp_pgno, NUMKEYS(mp));
2123                 assert(NUMKEYS(mp) > 1);
2124                 DPRINTF("found index 0 to page %lu", NODEPGNO(NODEPTR(mp, 0)));
2125
2126                 if (key == NULL)        /* Initialize cursor to first page. */
2127                         mc->mc_ki[mc->mc_top] = 0;
2128                 else if (key->mv_size > MAXKEYSIZE && key->mv_data == NULL) {
2129                                                         /* cursor to last page */
2130                         mc->mc_ki[mc->mc_top] = NUMKEYS(mp)-1;
2131                 } else {
2132                         int      exact;
2133                         node = mdb_search_node(mc, key, &exact);
2134                         if (node == NULL)
2135                                 mc->mc_ki[mc->mc_top] = NUMKEYS(mp) - 1;
2136                         else if (!exact) {
2137                                 assert(mc->mc_ki[mc->mc_top] > 0);
2138                                 mc->mc_ki[mc->mc_top]--;
2139                         }
2140                 }
2141
2142                 if (key)
2143                         DPRINTF("following index %u for key [%s]",
2144                             mc->mc_ki[mc->mc_top], DKEY(key));
2145                 assert(mc->mc_ki[mc->mc_top] < NUMKEYS(mp));
2146                 node = NODEPTR(mp, mc->mc_ki[mc->mc_top]);
2147
2148                 if ((rc = mdb_get_page(mc->mc_txn, NODEPGNO(node), &mp)))
2149                         return rc;
2150
2151                 if ((rc = cursor_push_page(mc, mp)))
2152                         return rc;
2153
2154                 if (modify) {
2155                         if ((rc = mdb_touch(mc)) != 0)
2156                                 return rc;
2157                         mp = mc->mc_pg[mc->mc_top];
2158                 }
2159         }
2160
2161         if (!IS_LEAF(mp)) {
2162                 DPRINTF("internal error, index points to a %02X page!?",
2163                     mp->mp_flags);
2164                 return MDB_CORRUPTED;
2165         }
2166
2167         DPRINTF("found leaf page %lu for key [%s]", mp->mp_pgno,
2168             key ? DKEY(key) : NULL);
2169
2170         return MDB_SUCCESS;
2171 }
2172
2173 /* Search for the page a given key should be in.
2174  * Pushes parent pages on the cursor stack.
2175  * If key is NULL, search for the lowest page (used by mdb_cursor_first).
2176  * If modify is true, visited pages are updated with new page numbers.
2177  */
2178 static int
2179 mdb_search_page(MDB_cursor *mc, MDB_val *key, int modify)
2180 {
2181         int              rc;
2182         pgno_t           root;
2183
2184         /* Make sure the txn is still viable, then find the root from
2185          * the txn's db table.
2186          */
2187         if (F_ISSET(mc->mc_txn->mt_flags, MDB_TXN_ERROR)) {
2188                 DPUTS("transaction has failed, must abort");
2189                 return EINVAL;
2190         } else
2191                 root = mc->mc_txn->mt_dbs[mc->mc_dbi].md_root;
2192
2193         if (root == P_INVALID) {                /* Tree is empty. */
2194                 DPUTS("tree is empty");
2195                 return MDB_NOTFOUND;
2196         }
2197
2198         if ((rc = mdb_get_page(mc->mc_txn, root, &mc->mc_pg[0])))
2199                 return rc;
2200
2201         mc->mc_snum = 1;
2202         mc->mc_top = 0;
2203
2204         DPRINTF("db %u root page %lu has flags 0x%X",
2205                 mc->mc_dbi, root, mc->mc_pg[0]->mp_flags);
2206
2207         if (modify) {
2208                 /* For sub-databases, update main root first */
2209                 if (mc->mc_dbi > MAIN_DBI && !mc->mc_txn->mt_dbxs[mc->mc_dbi].md_dirty) {
2210                         MDB_cursor mc2;
2211                         mc2.mc_txn = mc->mc_txn;
2212                         mc2.mc_dbi = MAIN_DBI;
2213                         rc = mdb_search_page(&mc2, &mc->mc_txn->mt_dbxs[mc->mc_dbi].md_name, 1);
2214                         if (rc)
2215                                 return rc;
2216                         mc->mc_txn->mt_dbxs[mc->mc_dbi].md_dirty = 1;
2217                 }
2218                 if (!F_ISSET(mc->mc_pg[0]->mp_flags, P_DIRTY)) {
2219                         if ((rc = mdb_touch(mc)))
2220                                 return rc;
2221                         mc->mc_txn->mt_dbs[mc->mc_dbi].md_root = mc->mc_pg[0]->mp_pgno;
2222                 }
2223         }
2224
2225         return mdb_search_page_root(mc, key, modify);
2226 }
2227
2228 static int
2229 mdb_read_data(MDB_txn *txn, MDB_node *leaf, MDB_val *data)
2230 {
2231         MDB_page        *omp;           /* overflow mpage */
2232         pgno_t           pgno;
2233         int rc;
2234
2235         if (!F_ISSET(leaf->mn_flags, F_BIGDATA)) {
2236                 data->mv_size = NODEDSZ(leaf);
2237                 data->mv_data = NODEDATA(leaf);
2238                 return MDB_SUCCESS;
2239         }
2240
2241         /* Read overflow data.
2242          */
2243         data->mv_size = NODEDSZ(leaf);
2244         memcpy(&pgno, NODEDATA(leaf), sizeof(pgno));
2245         if ((rc = mdb_get_page(txn, pgno, &omp))) {
2246                 DPRINTF("read overflow page %lu failed", pgno);
2247                 return rc;
2248         }
2249         data->mv_data = METADATA(omp);
2250
2251         return MDB_SUCCESS;
2252 }
2253
2254 int
2255 mdb_get(MDB_txn *txn, MDB_dbi dbi,
2256     MDB_val *key, MDB_val *data)
2257 {
2258         MDB_cursor      mc;
2259         MDB_xcursor     mx;
2260         int exact = 0;
2261         DKBUF;
2262
2263         assert(key);
2264         assert(data);
2265         DPRINTF("===> get db %u key [%s]", dbi, DKEY(key));
2266
2267         if (txn == NULL || !dbi || dbi >= txn->mt_numdbs)
2268                 return EINVAL;
2269
2270         if (key->mv_size == 0 || key->mv_size > MAXKEYSIZE) {
2271                 return EINVAL;
2272         }
2273
2274         mc.mc_txn = txn;
2275         mc.mc_dbi = dbi;
2276         mc.mc_flags = 0;
2277         if (txn->mt_dbs[dbi].md_flags & MDB_DUPSORT) {
2278                 mc.mc_xcursor = &mx;
2279                 mdb_xcursor_init0(&mc);
2280         } else {
2281                 mc.mc_xcursor = NULL;
2282         }
2283         return mdb_cursor_set(&mc, key, data, MDB_SET, &exact);
2284 }
2285
2286 static int
2287 mdb_sibling(MDB_cursor *mc, int move_right)
2288 {
2289         int              rc;
2290         unsigned int    ptop;
2291         MDB_node        *indx;
2292         MDB_page        *mp;
2293
2294         if (mc->mc_snum < 2) {
2295                 return MDB_NOTFOUND;            /* root has no siblings */
2296         }
2297         ptop = mc->mc_top-1;
2298
2299         DPRINTF("parent page is page %lu, index %u",
2300                 mc->mc_pg[ptop]->mp_pgno, mc->mc_ki[ptop]);
2301
2302         cursor_pop_page(mc);
2303         if (move_right ? (mc->mc_ki[ptop] + 1u >= NUMKEYS(mc->mc_pg[ptop]))
2304                        : (mc->mc_ki[ptop] == 0)) {
2305                 DPRINTF("no more keys left, moving to %s sibling",
2306                     move_right ? "right" : "left");
2307                 if ((rc = mdb_sibling(mc, move_right)) != MDB_SUCCESS)
2308                         return rc;
2309         } else {
2310                 if (move_right)
2311                         mc->mc_ki[ptop]++;
2312                 else
2313                         mc->mc_ki[ptop]--;
2314                 DPRINTF("just moving to %s index key %u",
2315                     move_right ? "right" : "left", mc->mc_ki[ptop]);
2316         }
2317         assert(IS_BRANCH(mc->mc_pg[ptop]));
2318
2319         indx = NODEPTR(mc->mc_pg[ptop], mc->mc_ki[ptop]);
2320         if ((rc = mdb_get_page(mc->mc_txn, NODEPGNO(indx), &mp)))
2321                 return rc;;
2322
2323         cursor_push_page(mc, mp);
2324
2325         return MDB_SUCCESS;
2326 }
2327
2328 static int
2329 mdb_cursor_next(MDB_cursor *mc, MDB_val *key, MDB_val *data, MDB_cursor_op op)
2330 {
2331         MDB_page        *mp;
2332         MDB_node        *leaf;
2333         int rc;
2334
2335         if (mc->mc_flags & C_EOF) {
2336                 return MDB_NOTFOUND;
2337         }
2338
2339         assert(mc->mc_flags & C_INITIALIZED);
2340
2341         mp = mc->mc_pg[mc->mc_top];
2342
2343         if (mc->mc_txn->mt_dbs[mc->mc_dbi].md_flags & MDB_DUPSORT) {
2344                 leaf = NODEPTR(mp, mc->mc_ki[mc->mc_top]);
2345                 if (F_ISSET(leaf->mn_flags, F_DUPDATA)) {
2346                         if (op == MDB_NEXT || op == MDB_NEXT_DUP) {
2347                                 rc = mdb_cursor_next(&mc->mc_xcursor->mx_cursor, data, NULL, MDB_NEXT);
2348                                 if (op != MDB_NEXT || rc == MDB_SUCCESS)
2349                                         return rc;
2350                         }
2351                 } else {
2352                         mc->mc_xcursor->mx_cursor.mc_flags = 0;
2353                         if (op == MDB_NEXT_DUP)
2354                                 return MDB_NOTFOUND;
2355                 }
2356         }
2357
2358         DPRINTF("cursor_next: top page is %lu in cursor %p", mp->mp_pgno, (void *) mc);
2359
2360         if (mc->mc_ki[mc->mc_top] + 1u >= NUMKEYS(mp)) {
2361                 DPUTS("=====> move to next sibling page");
2362                 if (mdb_sibling(mc, 1) != MDB_SUCCESS) {
2363                         mc->mc_flags |= C_EOF;
2364                         return MDB_NOTFOUND;
2365                 }
2366                 mp = mc->mc_pg[mc->mc_top];
2367                 DPRINTF("next page is %lu, key index %u", mp->mp_pgno, mc->mc_ki[mc->mc_top]);
2368         } else
2369                 mc->mc_ki[mc->mc_top]++;
2370
2371         DPRINTF("==> cursor points to page %lu with %u keys, key index %u",
2372             mp->mp_pgno, NUMKEYS(mp), mc->mc_ki[mc->mc_top]);
2373
2374         if (IS_LEAF2(mp)) {
2375                 key->mv_size = mc->mc_txn->mt_dbs[mc->mc_dbi].md_pad;
2376                 key->mv_data = LEAF2KEY(mp, mc->mc_ki[mc->mc_top], key->mv_size);
2377                 return MDB_SUCCESS;
2378         }
2379
2380         assert(IS_LEAF(mp));
2381         leaf = NODEPTR(mp, mc->mc_ki[mc->mc_top]);
2382
2383         if (F_ISSET(leaf->mn_flags, F_DUPDATA)) {
2384                 mdb_xcursor_init1(mc, leaf);
2385         }
2386         if (data) {
2387                 if ((rc = mdb_read_data(mc->mc_txn, leaf, data) != MDB_SUCCESS))
2388                         return rc;
2389
2390                 if (F_ISSET(leaf->mn_flags, F_DUPDATA)) {
2391                         rc = mdb_cursor_first(&mc->mc_xcursor->mx_cursor, data, NULL);
2392                         if (rc != MDB_SUCCESS)
2393                                 return rc;
2394                 }
2395         }
2396
2397         MDB_SET_KEY(leaf, key);
2398         return MDB_SUCCESS;
2399 }
2400
2401 static int
2402 mdb_cursor_prev(MDB_cursor *mc, MDB_val *key, MDB_val *data, MDB_cursor_op op)
2403 {
2404         MDB_page        *mp;
2405         MDB_node        *leaf;
2406         int rc;
2407
2408         assert(mc->mc_flags & C_INITIALIZED);
2409
2410         mp = mc->mc_pg[mc->mc_top];
2411
2412         if (mc->mc_txn->mt_dbs[mc->mc_dbi].md_flags & MDB_DUPSORT) {
2413                 leaf = NODEPTR(mp, mc->mc_ki[mc->mc_top]);
2414                 if (op == MDB_PREV || op == MDB_PREV_DUP) {
2415                         if (F_ISSET(leaf->mn_flags, F_DUPDATA)) {
2416                                 rc = mdb_cursor_prev(&mc->mc_xcursor->mx_cursor, data, NULL, MDB_PREV);
2417                                 if (op != MDB_PREV || rc == MDB_SUCCESS)
2418                                         return rc;
2419                         } else {
2420                                 mc->mc_xcursor->mx_cursor.mc_flags = 0;
2421                                 if (op == MDB_PREV_DUP)
2422                                         return MDB_NOTFOUND;
2423                         }
2424                 }
2425         }
2426
2427         DPRINTF("cursor_prev: top page is %lu in cursor %p", mp->mp_pgno, (void *) mc);
2428
2429         if (mc->mc_ki[mc->mc_top] == 0)  {
2430                 DPUTS("=====> move to prev sibling page");
2431                 if (mdb_sibling(mc, 0) != MDB_SUCCESS) {
2432                         mc->mc_flags &= ~C_INITIALIZED;
2433                         return MDB_NOTFOUND;
2434                 }
2435                 mp = mc->mc_pg[mc->mc_top];
2436                 mc->mc_ki[mc->mc_top] = NUMKEYS(mp) - 1;
2437                 DPRINTF("prev page is %lu, key index %u", mp->mp_pgno, mc->mc_ki[mc->mc_top]);
2438         } else
2439                 mc->mc_ki[mc->mc_top]--;
2440
2441         mc->mc_flags &= ~C_EOF;
2442
2443         DPRINTF("==> cursor points to page %lu with %u keys, key index %u",
2444             mp->mp_pgno, NUMKEYS(mp), mc->mc_ki[mc->mc_top]);
2445
2446         if (IS_LEAF2(mp)) {
2447                 key->mv_size = mc->mc_txn->mt_dbs[mc->mc_dbi].md_pad;
2448                 key->mv_data = LEAF2KEY(mp, mc->mc_ki[mc->mc_top], key->mv_size);
2449                 return MDB_SUCCESS;
2450         }
2451
2452         assert(IS_LEAF(mp));
2453         leaf = NODEPTR(mp, mc->mc_ki[mc->mc_top]);
2454
2455         if (F_ISSET(leaf->mn_flags, F_DUPDATA)) {
2456                 mdb_xcursor_init1(mc, leaf);
2457         }
2458         if (data) {
2459                 if ((rc = mdb_read_data(mc->mc_txn, leaf, data) != MDB_SUCCESS))
2460                         return rc;
2461
2462                 if (F_ISSET(leaf->mn_flags, F_DUPDATA)) {
2463                         rc = mdb_cursor_last(&mc->mc_xcursor->mx_cursor, data, NULL);
2464                         if (rc != MDB_SUCCESS)
2465                                 return rc;
2466                 }
2467         }
2468
2469         MDB_SET_KEY(leaf, key);
2470         return MDB_SUCCESS;
2471 }
2472
2473 static int
2474 mdb_cursor_set(MDB_cursor *mc, MDB_val *key, MDB_val *data,
2475     MDB_cursor_op op, int *exactp)
2476 {
2477         int              rc;
2478         MDB_node        *leaf;
2479         DKBUF;
2480
2481         assert(mc);
2482         assert(key);
2483         assert(key->mv_size > 0);
2484
2485         /* See if we're already on the right page */
2486         if (mc->mc_flags & C_INITIALIZED) {
2487                 MDB_val nodekey;
2488
2489                 if (mc->mc_pg[mc->mc_top]->mp_flags & P_LEAF2) {
2490                         nodekey.mv_size = mc->mc_txn->mt_dbs[mc->mc_dbi].md_pad;
2491                         nodekey.mv_data = LEAF2KEY(mc->mc_pg[mc->mc_top], 0, nodekey.mv_size);
2492                 } else {
2493                         leaf = NODEPTR(mc->mc_pg[mc->mc_top], 0);
2494                         MDB_SET_KEY(leaf, &nodekey);
2495                 }
2496                 rc = mc->mc_txn->mt_dbxs[mc->mc_dbi].md_cmp(key, &nodekey);
2497                 if (rc == 0) {
2498                         /* Probably happens rarely, but first node on the page
2499                          * was the one we wanted.
2500                          */
2501                         mc->mc_ki[mc->mc_top] = 0;
2502 set1:
2503                         if (exactp)
2504                                 *exactp = 1;
2505                         rc = 0;
2506                         leaf = NODEPTR(mc->mc_pg[mc->mc_top], mc->mc_ki[mc->mc_top]);
2507                         goto set3;
2508                 }
2509                 if (rc > 0) {
2510                         unsigned int i;
2511                         if (NUMKEYS(mc->mc_pg[mc->mc_top]) > 1) {
2512                                 if (mc->mc_pg[mc->mc_top]->mp_flags & P_LEAF2) {
2513                                         nodekey.mv_data = LEAF2KEY(mc->mc_pg[mc->mc_top],
2514                                                  NUMKEYS(mc->mc_pg[mc->mc_top])-1, nodekey.mv_size);
2515                                 } else {
2516                                         leaf = NODEPTR(mc->mc_pg[mc->mc_top], NUMKEYS(mc->mc_pg[mc->mc_top])-1);
2517                                         MDB_SET_KEY(leaf, &nodekey);
2518                                 }
2519                                 rc = mc->mc_txn->mt_dbxs[mc->mc_dbi].md_cmp(key, &nodekey);
2520                                 if (rc == 0) {
2521                                         /* last node was the one we wanted */
2522                                         mc->mc_ki[mc->mc_top] = NUMKEYS(mc->mc_pg[mc->mc_top])-1;
2523                                         goto set1;
2524                                 }
2525                                 if (rc < 0) {
2526                                         /* This is definitely the right page, skip search_page */
2527                                         rc = 0;
2528                                         goto set2;
2529                                 }
2530                         }
2531                         /* If any parents have right-sibs, search.
2532                          * Otherwise, there's nothing further.
2533                          */
2534                         for (i=0; i<mc->mc_top; i++)
2535                                 if (mc->mc_ki[i] <
2536                                         NUMKEYS(mc->mc_pg[i])-1)
2537                                         break;
2538                         if (i == mc->mc_top) {
2539                                 /* There are no other pages */
2540                                 mc->mc_ki[mc->mc_top] = NUMKEYS(mc->mc_pg[mc->mc_top]);
2541                                 return MDB_NOTFOUND;
2542                         }
2543                 }
2544         }
2545
2546         rc = mdb_search_page(mc, key, 0);
2547         if (rc != MDB_SUCCESS)
2548                 return rc;
2549
2550         assert(IS_LEAF(mc->mc_pg[mc->mc_top]));
2551
2552 set2:
2553         leaf = mdb_search_node(mc, key, exactp);
2554         if (exactp != NULL && !*exactp) {
2555                 /* MDB_SET specified and not an exact match. */
2556                 return MDB_NOTFOUND;
2557         }
2558
2559         if (leaf == NULL) {
2560                 DPUTS("===> inexact leaf not found, goto sibling");
2561                 if ((rc = mdb_sibling(mc, 1)) != MDB_SUCCESS)
2562                         return rc;              /* no entries matched */
2563                 mc->mc_ki[mc->mc_top] = 0;
2564                 assert(IS_LEAF(mc->mc_pg[mc->mc_top]));
2565                 leaf = NODEPTR(mc->mc_pg[mc->mc_top], 0);
2566         }
2567
2568 set3:
2569         mc->mc_flags |= C_INITIALIZED;
2570         mc->mc_flags &= ~C_EOF;
2571
2572         if (IS_LEAF2(mc->mc_pg[mc->mc_top])) {
2573                 key->mv_size = mc->mc_txn->mt_dbs[mc->mc_dbi].md_pad;
2574                 key->mv_data = LEAF2KEY(mc->mc_pg[mc->mc_top], mc->mc_ki[mc->mc_top], key->mv_size);
2575                 return MDB_SUCCESS;
2576         }
2577
2578         if (F_ISSET(leaf->mn_flags, F_DUPDATA)) {
2579                 mdb_xcursor_init1(mc, leaf);
2580         }
2581         if (data) {
2582                 if (F_ISSET(leaf->mn_flags, F_DUPDATA)) {
2583                         if (op == MDB_SET || op == MDB_SET_RANGE) {
2584                                 rc = mdb_cursor_first(&mc->mc_xcursor->mx_cursor, data, NULL);
2585                         } else {
2586                                 int ex2, *ex2p;
2587                                 if (op == MDB_GET_BOTH) {
2588                                         ex2p = &ex2;
2589                                         ex2 = 0;
2590                                 } else {
2591                                         ex2p = NULL;
2592                                 }
2593                                 rc = mdb_cursor_set(&mc->mc_xcursor->mx_cursor, data, NULL, MDB_SET_RANGE, ex2p);
2594                                 if (rc != MDB_SUCCESS)
2595                                         return rc;
2596                         }
2597                 } else if (op == MDB_GET_BOTH || op == MDB_GET_BOTH_RANGE) {
2598                         MDB_val d2;
2599                         if ((rc = mdb_read_data(mc->mc_txn, leaf, &d2)) != MDB_SUCCESS)
2600                                 return rc;
2601                         rc = mc->mc_txn->mt_dbxs[mc->mc_dbi].md_dcmp(data, &d2);
2602                         if (rc) {
2603                                 if (op == MDB_GET_BOTH || rc > 0)
2604                                         return MDB_NOTFOUND;
2605                         }
2606
2607                 } else {
2608                         if ((rc = mdb_read_data(mc->mc_txn, leaf, data)) != MDB_SUCCESS)
2609                                 return rc;
2610                 }
2611         }
2612
2613         /* The key already matches in all other cases */
2614         if (op == MDB_SET_RANGE)
2615                 MDB_SET_KEY(leaf, key);
2616         DPRINTF("==> cursor placed on key [%s]", DKEY(key));
2617
2618         return rc;
2619 }
2620
2621 static int
2622 mdb_cursor_first(MDB_cursor *mc, MDB_val *key, MDB_val *data)
2623 {
2624         int              rc;
2625         MDB_node        *leaf;
2626
2627         rc = mdb_search_page(mc, NULL, 0);
2628         if (rc != MDB_SUCCESS)
2629                 return rc;
2630         assert(IS_LEAF(mc->mc_pg[mc->mc_top]));
2631
2632         leaf = NODEPTR(mc->mc_pg[mc->mc_top], 0);
2633         mc->mc_flags |= C_INITIALIZED;
2634         mc->mc_flags &= ~C_EOF;
2635
2636         if (IS_LEAF2(mc->mc_pg[mc->mc_top])) {
2637                 key->mv_size = mc->mc_txn->mt_dbs[mc->mc_dbi].md_pad;
2638                 key->mv_data = LEAF2KEY(mc->mc_pg[mc->mc_top], 0, key->mv_size);
2639                 return MDB_SUCCESS;
2640         }
2641
2642         if (data) {
2643                 if (F_ISSET(leaf->mn_flags, F_DUPDATA)) {
2644                         mdb_xcursor_init1(mc, leaf);
2645                         rc = mdb_cursor_first(&mc->mc_xcursor->mx_cursor, data, NULL);
2646                         if (rc)
2647                                 return rc;
2648                 } else {
2649                         if (mc->mc_xcursor)
2650                                 mc->mc_xcursor->mx_cursor.mc_flags = 0;
2651                         if ((rc = mdb_read_data(mc->mc_txn, leaf, data)) != MDB_SUCCESS)
2652                                 return rc;
2653                 }
2654         }
2655         MDB_SET_KEY(leaf, key);
2656         return MDB_SUCCESS;
2657 }
2658
2659 static int
2660 mdb_cursor_last(MDB_cursor *mc, MDB_val *key, MDB_val *data)
2661 {
2662         int              rc;
2663         MDB_node        *leaf;
2664         MDB_val lkey;
2665
2666         lkey.mv_size = MAXKEYSIZE+1;
2667         lkey.mv_data = NULL;
2668
2669         rc = mdb_search_page(mc, &lkey, 0);
2670         if (rc != MDB_SUCCESS)
2671                 return rc;
2672         assert(IS_LEAF(mc->mc_pg[mc->mc_top]));
2673
2674         leaf = NODEPTR(mc->mc_pg[mc->mc_top], NUMKEYS(mc->mc_pg[mc->mc_top])-1);
2675         mc->mc_flags |= C_INITIALIZED;
2676         mc->mc_flags &= ~C_EOF;
2677
2678         mc->mc_ki[mc->mc_top] = NUMKEYS(mc->mc_pg[mc->mc_top]) - 1;
2679
2680         if (IS_LEAF2(mc->mc_pg[mc->mc_top])) {
2681                 key->mv_size = mc->mc_txn->mt_dbs[mc->mc_dbi].md_pad;
2682                 key->mv_data = LEAF2KEY(mc->mc_pg[mc->mc_top], mc->mc_ki[mc->mc_top], key->mv_size);
2683                 return MDB_SUCCESS;
2684         }
2685
2686         if (data) {
2687                 if (F_ISSET(leaf->mn_flags, F_DUPDATA)) {
2688                         mdb_xcursor_init1(mc, leaf);
2689                         rc = mdb_cursor_last(&mc->mc_xcursor->mx_cursor, data, NULL);
2690                         if (rc)
2691                                 return rc;
2692                 } else {
2693                         if ((rc = mdb_read_data(mc->mc_txn, leaf, data)) != MDB_SUCCESS)
2694                                 return rc;
2695                 }
2696         }
2697
2698         MDB_SET_KEY(leaf, key);
2699         return MDB_SUCCESS;
2700 }
2701
2702 int
2703 mdb_cursor_get(MDB_cursor *mc, MDB_val *key, MDB_val *data,
2704     MDB_cursor_op op)
2705 {
2706         int              rc;
2707         int              exact = 0;
2708
2709         assert(mc);
2710
2711         switch (op) {
2712         case MDB_GET_BOTH:
2713         case MDB_GET_BOTH_RANGE:
2714                 if (data == NULL || mc->mc_xcursor == NULL) {
2715                         rc = EINVAL;
2716                         break;
2717                 }
2718                 /* FALLTHRU */
2719         case MDB_SET:
2720         case MDB_SET_RANGE:
2721                 if (key == NULL || key->mv_size == 0 || key->mv_size > MAXKEYSIZE) {
2722                         rc = EINVAL;
2723                 } else if (op == MDB_SET_RANGE)
2724                         rc = mdb_cursor_set(mc, key, data, op, NULL);
2725                 else
2726                         rc = mdb_cursor_set(mc, key, data, op, &exact);
2727                 break;
2728         case MDB_GET_MULTIPLE:
2729                 if (data == NULL ||
2730                         !(mc->mc_txn->mt_dbs[mc->mc_dbi].md_flags & MDB_DUPFIXED) ||
2731                         !(mc->mc_flags & C_INITIALIZED)) {
2732                         rc = EINVAL;
2733                         break;
2734                 }
2735                 rc = MDB_SUCCESS;
2736                 if (!(mc->mc_xcursor->mx_cursor.mc_flags & C_INITIALIZED) ||
2737                         (mc->mc_xcursor->mx_cursor.mc_flags & C_EOF))
2738                         break;
2739                 goto fetchm;
2740         case MDB_NEXT_MULTIPLE:
2741                 if (data == NULL ||
2742                         !(mc->mc_txn->mt_dbs[mc->mc_dbi].md_flags & MDB_DUPFIXED)) {
2743                         rc = EINVAL;
2744                         break;
2745                 }
2746                 if (!(mc->mc_flags & C_INITIALIZED))
2747                         rc = mdb_cursor_first(mc, key, data);
2748                 else
2749                         rc = mdb_cursor_next(mc, key, data, MDB_NEXT_DUP);
2750                 if (rc == MDB_SUCCESS) {
2751                         if (mc->mc_xcursor->mx_cursor.mc_flags & C_INITIALIZED) {
2752                                 MDB_cursor *mx;
2753 fetchm:
2754                                 mx = &mc->mc_xcursor->mx_cursor;
2755                                 data->mv_size = NUMKEYS(mx->mc_pg[mx->mc_top]) *
2756                                         mx->mc_txn->mt_dbs[mx->mc_dbi].md_pad;
2757                                 data->mv_data = METADATA(mx->mc_pg[mx->mc_top]);
2758                                 mx->mc_ki[mx->mc_top] = NUMKEYS(mx->mc_pg[mx->mc_top])-1;
2759                         } else {
2760                                 rc = MDB_NOTFOUND;
2761                         }
2762                 }
2763                 break;
2764         case MDB_NEXT:
2765         case MDB_NEXT_DUP:
2766         case MDB_NEXT_NODUP:
2767                 if (!(mc->mc_flags & C_INITIALIZED))
2768                         rc = mdb_cursor_first(mc, key, data);
2769                 else
2770                         rc = mdb_cursor_next(mc, key, data, op);
2771                 break;
2772         case MDB_PREV:
2773         case MDB_PREV_DUP:
2774         case MDB_PREV_NODUP:
2775                 if (!(mc->mc_flags & C_INITIALIZED) || (mc->mc_flags & C_EOF))
2776                         rc = mdb_cursor_last(mc, key, data);
2777                 else
2778                         rc = mdb_cursor_prev(mc, key, data, op);
2779                 break;
2780         case MDB_FIRST:
2781                 rc = mdb_cursor_first(mc, key, data);
2782                 break;
2783         case MDB_FIRST_DUP:
2784                 if (data == NULL ||
2785                         !(mc->mc_txn->mt_dbs[mc->mc_dbi].md_flags & MDB_DUPSORT) ||
2786                         !(mc->mc_flags & C_INITIALIZED) ||
2787                         !(mc->mc_xcursor->mx_cursor.mc_flags & C_INITIALIZED)) {
2788                         rc = EINVAL;
2789                         break;
2790                 }
2791                 rc = mdb_cursor_first(&mc->mc_xcursor->mx_cursor, data, NULL);
2792                 break;
2793         case MDB_LAST:
2794                 rc = mdb_cursor_last(mc, key, data);
2795                 break;
2796         case MDB_LAST_DUP:
2797                 if (data == NULL ||
2798                         !(mc->mc_txn->mt_dbs[mc->mc_dbi].md_flags & MDB_DUPSORT) ||
2799                         !(mc->mc_flags & C_INITIALIZED) ||
2800                         !(mc->mc_xcursor->mx_cursor.mc_flags & C_INITIALIZED)) {
2801                         rc = EINVAL;
2802                         break;
2803                 }
2804                 rc = mdb_cursor_last(&mc->mc_xcursor->mx_cursor, data, NULL);
2805                 break;
2806         default:
2807                 DPRINTF("unhandled/unimplemented cursor operation %u", op);
2808                 rc = EINVAL;
2809                 break;
2810         }
2811
2812         return rc;
2813 }
2814
2815 static int
2816 mdb_cursor_touch(MDB_cursor *mc)
2817 {
2818         int rc;
2819
2820         if (mc->mc_dbi > MAIN_DBI && !mc->mc_txn->mt_dbxs[mc->mc_dbi].md_dirty) {
2821                 MDB_cursor mc2;
2822                 mc2.mc_txn = mc->mc_txn;
2823                 mc2.mc_dbi = MAIN_DBI;
2824                 rc = mdb_search_page(&mc2, &mc->mc_txn->mt_dbxs[mc->mc_dbi].md_name, 1);
2825                 if (rc) return rc;
2826                 mc->mc_txn->mt_dbxs[mc->mc_dbi].md_dirty = 1;
2827         }
2828         for (mc->mc_top = 0; mc->mc_top < mc->mc_snum; mc->mc_top++) {
2829                 if (!F_ISSET(mc->mc_pg[mc->mc_top]->mp_flags, P_DIRTY)) {
2830                         rc = mdb_touch(mc);
2831                         if (rc) return rc;
2832                         if (!mc->mc_top) {
2833                                 mc->mc_txn->mt_dbs[mc->mc_dbi].md_root =
2834                                         mc->mc_pg[mc->mc_top]->mp_pgno;
2835                         }
2836                 }
2837         }
2838         mc->mc_top = mc->mc_snum-1;
2839         return MDB_SUCCESS;
2840 }
2841
2842 int
2843 mdb_cursor_put(MDB_cursor *mc, MDB_val *key, MDB_val *data,
2844     unsigned int flags)
2845 {
2846         MDB_node        *leaf;
2847         MDB_val xdata, *rdata, dkey;
2848         MDB_db dummy;
2849         char dbuf[PAGESIZE];
2850         int do_sub = 0;
2851         size_t nsize;
2852         DKBUF;
2853         int rc, rc2;
2854
2855         if (F_ISSET(mc->mc_txn->mt_flags, MDB_TXN_RDONLY))
2856                 return EACCES;
2857
2858         DPRINTF("==> put db %u key [%s], size %zu, data size %zu",
2859                 mc->mc_dbi, DKEY(key), key->mv_size, data->mv_size);
2860
2861         dkey.mv_size = 0;
2862
2863         if (flags == MDB_CURRENT) {
2864                 if (!(mc->mc_flags & C_INITIALIZED))
2865                         return EINVAL;
2866                 rc = MDB_SUCCESS;
2867         } else if (mc->mc_txn->mt_dbs[mc->mc_dbi].md_root == P_INVALID) {
2868                 MDB_page *np;
2869                 /* new database, write a root leaf page */
2870                 DPUTS("allocating new root leaf page");
2871                 if ((np = mdb_new_page(mc, P_LEAF, 1)) == NULL) {
2872                         return ENOMEM;
2873                 }
2874                 mc->mc_snum = 0;
2875                 cursor_push_page(mc, np);
2876                 mc->mc_txn->mt_dbs[mc->mc_dbi].md_root = np->mp_pgno;
2877                 mc->mc_txn->mt_dbs[mc->mc_dbi].md_depth++;
2878                 mc->mc_txn->mt_dbxs[mc->mc_dbi].md_dirty = 1;
2879                 if ((mc->mc_txn->mt_dbs[mc->mc_dbi].md_flags & (MDB_DUPSORT|MDB_DUPFIXED))
2880                         == MDB_DUPFIXED)
2881                         np->mp_flags |= P_LEAF2;
2882                 mc->mc_flags |= C_INITIALIZED;
2883                 rc = MDB_NOTFOUND;
2884                 goto top;
2885         } else {
2886                 int exact = 0;
2887                 rc = mdb_cursor_set(mc, key, NULL, MDB_SET, &exact);
2888                 if (flags == MDB_NOOVERWRITE && rc == 0) {
2889                         DPRINTF("duplicate key [%s]", DKEY(key));
2890                         return MDB_KEYEXIST;
2891                 }
2892                 if (rc && rc != MDB_NOTFOUND)
2893                         return rc;
2894         }
2895
2896         /* Cursor is positioned, now make sure all pages are writable */
2897         rc2 = mdb_cursor_touch(mc);
2898         if (rc2) return rc2;
2899
2900 top:
2901         /* The key already exists */
2902         if (rc == MDB_SUCCESS) {
2903                 /* there's only a key anyway, so this is a no-op */
2904                 if (IS_LEAF2(mc->mc_pg[mc->mc_top])) {
2905                         unsigned int ksize = mc->mc_txn->mt_dbs[mc->mc_dbi].md_pad;
2906                         if (key->mv_size != ksize)
2907                                 return EINVAL;
2908                         if (flags == MDB_CURRENT) {
2909                                 char *ptr = LEAF2KEY(mc->mc_pg[mc->mc_top], mc->mc_ki[mc->mc_top], ksize);
2910                                 memcpy(ptr, key->mv_data, ksize);
2911                         }
2912                         return MDB_SUCCESS;
2913                 }
2914
2915                 leaf = NODEPTR(mc->mc_pg[mc->mc_top], mc->mc_ki[mc->mc_top]);
2916
2917                 /* DB has dups? */
2918                 if (F_ISSET(mc->mc_txn->mt_dbs[mc->mc_dbi].md_flags, MDB_DUPSORT)) {
2919                         /* Was a single item before, must convert now */
2920                         if (!F_ISSET(leaf->mn_flags, F_DUPDATA)) {
2921                                 dkey.mv_size = NODEDSZ(leaf);
2922                                 dkey.mv_data = dbuf;
2923                                 memcpy(dbuf, NODEDATA(leaf), dkey.mv_size);
2924                                 /* data matches, ignore it */
2925                                 if (!mdb_dcmp(mc->mc_txn, mc->mc_dbi, data, &dkey))
2926                                         return (flags == MDB_NODUPDATA) ? MDB_KEYEXIST : MDB_SUCCESS;
2927                                 memset(&dummy, 0, sizeof(dummy));
2928                                 if (mc->mc_txn->mt_dbs[mc->mc_dbi].md_flags & MDB_DUPFIXED) {
2929                                         dummy.md_pad = data->mv_size;
2930                                         dummy.md_flags = MDB_DUPFIXED;
2931                                         if (mc->mc_txn->mt_dbs[mc->mc_dbi].md_flags & MDB_INTEGERDUP)
2932                                                 dummy.md_flags |= MDB_INTEGERKEY;
2933                                 }
2934                                 dummy.md_root = P_INVALID;
2935                                 if (dkey.mv_size == sizeof(MDB_db)) {
2936                                         memcpy(NODEDATA(leaf), &dummy, sizeof(dummy));
2937                                         goto put_sub;
2938                                 }
2939                                 mdb_del_node(mc->mc_pg[mc->mc_top], mc->mc_ki[mc->mc_top], 0);
2940                                 do_sub = 1;
2941                                 rdata = &xdata;
2942                                 xdata.mv_size = sizeof(MDB_db);
2943                                 xdata.mv_data = &dummy;
2944                                 goto new_sub;
2945                         }
2946                         goto put_sub;
2947                 }
2948                 /* same size, just replace it */
2949                 if (!F_ISSET(leaf->mn_flags, F_BIGDATA) &&
2950                         NODEDSZ(leaf) == data->mv_size) {
2951                         memcpy(NODEDATA(leaf), data->mv_data, data->mv_size);
2952                         goto done;
2953                 }
2954                 mdb_del_node(mc->mc_pg[mc->mc_top], mc->mc_ki[mc->mc_top], 0);
2955         } else {
2956                 DPRINTF("inserting key at index %i", mc->mc_ki[mc->mc_top]);
2957         }
2958
2959         rdata = data;
2960
2961 new_sub:
2962         nsize = IS_LEAF2(mc->mc_pg[mc->mc_top]) ? key->mv_size : mdb_leaf_size(mc->mc_txn->mt_env, key, rdata);
2963         if (SIZELEFT(mc->mc_pg[mc->mc_top]) < nsize) {
2964                 rc = mdb_split(mc, key, rdata, P_INVALID);
2965         } else {
2966                 /* There is room already in this leaf page. */
2967                 rc = mdb_add_node(mc, mc->mc_ki[mc->mc_top], key, rdata, 0, 0);
2968         }
2969
2970         if (rc != MDB_SUCCESS)
2971                 mc->mc_txn->mt_flags |= MDB_TXN_ERROR;
2972         else {
2973                 /* Remember if we just added a subdatabase */
2974                 if (flags & F_SUBDATA) {
2975                         leaf = NODEPTR(mc->mc_pg[mc->mc_top], mc->mc_ki[mc->mc_top]);
2976                         leaf->mn_flags |= F_SUBDATA;
2977                 }
2978
2979                 /* Now store the actual data in the child DB. Note that we're
2980                  * storing the user data in the keys field, so there are strict
2981                  * size limits on dupdata. The actual data fields of the child
2982                  * DB are all zero size.
2983                  */
2984                 if (do_sub) {
2985                         leaf = NODEPTR(mc->mc_pg[mc->mc_top], mc->mc_ki[mc->mc_top]);
2986 put_sub:
2987                         if (flags == MDB_CURRENT)
2988                                 mdb_xcursor_init2(mc);
2989                         else
2990                                 mdb_xcursor_init1(mc, leaf);
2991                         xdata.mv_size = 0;
2992                         xdata.mv_data = "";
2993                         if (flags == MDB_NODUPDATA)
2994                                 flags = MDB_NOOVERWRITE;
2995                         /* converted, write the original data first */
2996                         if (dkey.mv_size) {
2997                                 rc = mdb_cursor_put(&mc->mc_xcursor->mx_cursor, &dkey, &xdata, flags);
2998                                 if (rc) return rc;
2999                                 leaf->mn_flags |= F_DUPDATA;
3000                         }
3001                         rc = mdb_cursor_put(&mc->mc_xcursor->mx_cursor, data, &xdata, flags);
3002                         mdb_xcursor_fini(mc);
3003                         memcpy(NODEDATA(leaf),
3004                                 &mc->mc_xcursor->mx_txn.mt_dbs[mc->mc_xcursor->mx_cursor.mc_dbi],
3005                                 sizeof(MDB_db));
3006                 }
3007                 mc->mc_txn->mt_dbs[mc->mc_dbi].md_entries++;
3008         }
3009 done:
3010         return rc;
3011 }
3012
3013 int
3014 mdb_cursor_del(MDB_cursor *mc, unsigned int flags)
3015 {
3016         MDB_node        *leaf;
3017         int rc;
3018
3019         if (F_ISSET(mc->mc_txn->mt_flags, MDB_TXN_RDONLY))
3020                 return EACCES;
3021
3022         if (!mc->mc_flags & C_INITIALIZED)
3023                 return EINVAL;
3024
3025         rc = mdb_cursor_touch(mc);
3026         if (rc) return rc;
3027
3028         leaf = NODEPTR(mc->mc_pg[mc->mc_top], mc->mc_ki[mc->mc_top]);
3029
3030         if (!IS_LEAF2(mc->mc_pg[mc->mc_top]) && F_ISSET(leaf->mn_flags, F_DUPDATA)) {
3031                 if (flags != MDB_NODUPDATA) {
3032                         mdb_xcursor_init2(mc);
3033                         rc = mdb_cursor_del(&mc->mc_xcursor->mx_cursor, 0);
3034                         mdb_xcursor_fini(mc);
3035                         /* If sub-DB still has entries, we're done */
3036                         if (mc->mc_xcursor->mx_txn.mt_dbs[mc->mc_xcursor->mx_cursor.mc_dbi].md_root
3037                                 != P_INVALID) {
3038                                 memcpy(NODEDATA(leaf),
3039                                         &mc->mc_xcursor->mx_txn.mt_dbs[mc->mc_xcursor->mx_cursor.mc_dbi],
3040                                         sizeof(MDB_db));
3041                                 mc->mc_txn->mt_dbs[mc->mc_dbi].md_entries--;
3042                                 return rc;
3043                         }
3044                         /* otherwise fall thru and delete the sub-DB */
3045                 }
3046
3047                 /* add all the child DB's pages to the free list */
3048                 rc = mdb_search_page(&mc->mc_xcursor->mx_cursor, NULL, 0);
3049                 if (rc == MDB_SUCCESS) {
3050                         MDB_node *ni;
3051                         MDB_cursor *mx;
3052                         unsigned int i;
3053
3054                         mx = &mc->mc_xcursor->mx_cursor;
3055                         mc->mc_txn->mt_dbs[mc->mc_dbi].md_entries -=
3056                                 mx->mc_txn->mt_dbs[mx->mc_dbi].md_entries;
3057
3058                         cursor_pop_page(mx);
3059                         if (mx->mc_snum) {
3060                                 while (mx->mc_snum > 1) {
3061                                         for (i=0; i<NUMKEYS(mx->mc_pg[mx->mc_top]); i++) {
3062                                                 pgno_t pg;
3063                                                 ni = NODEPTR(mx->mc_pg[mx->mc_top], i);
3064                                                 pg = NODEPGNO(ni);
3065                                                 /* free it */
3066                                                 mdb_midl_append(mc->mc_txn->mt_free_pgs, pg);
3067                                         }
3068                                         rc = mdb_sibling(mx, 1);
3069                                         if (rc) break;
3070                                 }
3071                         }
3072                         /* free it */
3073                         mdb_midl_append(mc->mc_txn->mt_free_pgs,
3074                                 mx->mc_txn->mt_dbs[mx->mc_dbi].md_root);
3075                 }
3076         }
3077
3078         return mdb_del0(mc, leaf);
3079 }
3080
3081 /* Allocate a page and initialize it
3082  */
3083 static MDB_page *
3084 mdb_new_page(MDB_cursor *mc, uint32_t flags, int num)
3085 {
3086         MDB_page        *np;
3087
3088         if ((np = mdb_alloc_page(mc, num)) == NULL)
3089                 return NULL;
3090         DPRINTF("allocated new mpage %lu, page size %u",
3091             np->mp_pgno, mc->mc_txn->mt_env->me_psize);
3092         np->mp_flags = flags | P_DIRTY;
3093         np->mp_lower = PAGEHDRSZ;
3094         np->mp_upper = mc->mc_txn->mt_env->me_psize;
3095
3096         if (IS_BRANCH(np))
3097                 mc->mc_txn->mt_dbs[mc->mc_dbi].md_branch_pages++;
3098         else if (IS_LEAF(np))
3099                 mc->mc_txn->mt_dbs[mc->mc_dbi].md_leaf_pages++;
3100         else if (IS_OVERFLOW(np)) {
3101                 mc->mc_txn->mt_dbs[mc->mc_dbi].md_overflow_pages += num;
3102                 np->mp_pages = num;
3103         }
3104
3105         return np;
3106 }
3107
3108 static size_t
3109 mdb_leaf_size(MDB_env *env, MDB_val *key, MDB_val *data)
3110 {
3111         size_t           sz;
3112
3113         sz = LEAFSIZE(key, data);
3114         if (data->mv_size >= env->me_psize / MDB_MINKEYS) {
3115                 /* put on overflow page */
3116                 sz -= data->mv_size - sizeof(pgno_t);
3117         }
3118         sz += sz & 1;
3119
3120         return sz + sizeof(indx_t);
3121 }
3122
3123 static size_t
3124 mdb_branch_size(MDB_env *env, MDB_val *key)
3125 {
3126         size_t           sz;
3127
3128         sz = INDXSIZE(key);
3129         if (sz >= env->me_psize / MDB_MINKEYS) {
3130                 /* put on overflow page */
3131                 /* not implemented */
3132                 /* sz -= key->size - sizeof(pgno_t); */
3133         }
3134
3135         return sz + sizeof(indx_t);
3136 }
3137
3138 static int
3139 mdb_add_node(MDB_cursor *mc, indx_t indx,
3140     MDB_val *key, MDB_val *data, pgno_t pgno, uint8_t flags)
3141 {
3142         unsigned int     i;
3143         size_t           node_size = NODESIZE;
3144         indx_t           ofs;
3145         MDB_node        *node;
3146         MDB_page        *mp = mc->mc_pg[mc->mc_top];
3147         MDB_page        *ofp = NULL;            /* overflow page */
3148         DKBUF;
3149
3150         assert(mp->mp_upper >= mp->mp_lower);
3151
3152         DPRINTF("add to %s page %lu index %i, data size %zu key size %zu [%s]",
3153             IS_LEAF(mp) ? "leaf" : "branch",
3154             mp->mp_pgno, indx, data ? data->mv_size : 0,
3155                 key ? key->mv_size : 0, key ? DKEY(key) : NULL);
3156
3157         if (IS_LEAF2(mp)) {
3158                 /* Move higher keys up one slot. */
3159                 int ksize = mc->mc_txn->mt_dbs[mc->mc_dbi].md_pad, dif;
3160                 char *ptr = LEAF2KEY(mp, indx, ksize);
3161                 dif = NUMKEYS(mp) - indx;
3162                 if (dif > 0)
3163                         memmove(ptr+ksize, ptr, dif*ksize);
3164                 /* insert new key */
3165                 memcpy(ptr, key->mv_data, ksize);
3166
3167                 /* Just using these for counting */
3168                 mp->mp_lower += sizeof(indx_t);
3169                 mp->mp_upper -= ksize - sizeof(indx_t);
3170                 return MDB_SUCCESS;
3171         }
3172
3173         if (key != NULL)
3174                 node_size += key->mv_size;
3175
3176         if (IS_LEAF(mp)) {
3177                 assert(data);
3178                 if (F_ISSET(flags, F_BIGDATA)) {
3179                         /* Data already on overflow page. */
3180                         node_size += sizeof(pgno_t);
3181                 } else if (data->mv_size >= mc->mc_txn->mt_env->me_psize / MDB_MINKEYS) {
3182                         int ovpages = OVPAGES(data->mv_size, mc->mc_txn->mt_env->me_psize);
3183                         /* Put data on overflow page. */
3184                         DPRINTF("data size is %zu, put on overflow page",
3185                             data->mv_size);
3186                         node_size += sizeof(pgno_t);
3187                         if ((ofp = mdb_new_page(mc, P_OVERFLOW, ovpages)) == NULL)
3188                                 return ENOMEM;
3189                         DPRINTF("allocated overflow page %lu", ofp->mp_pgno);
3190                         flags |= F_BIGDATA;
3191                 } else {
3192                         node_size += data->mv_size;
3193                 }
3194         }
3195         node_size += node_size & 1;
3196
3197         if (node_size + sizeof(indx_t) > SIZELEFT(mp)) {
3198                 DPRINTF("not enough room in page %lu, got %u ptrs",
3199                     mp->mp_pgno, NUMKEYS(mp));
3200                 DPRINTF("upper - lower = %u - %u = %u", mp->mp_upper, mp->mp_lower,
3201                     mp->mp_upper - mp->mp_lower);
3202                 DPRINTF("node size = %zu", node_size);
3203                 return ENOSPC;
3204         }
3205
3206         /* Move higher pointers up one slot. */
3207         for (i = NUMKEYS(mp); i > indx; i--)
3208                 mp->mp_ptrs[i] = mp->mp_ptrs[i - 1];
3209
3210         /* Adjust free space offsets. */
3211         ofs = mp->mp_upper - node_size;
3212         assert(ofs >= mp->mp_lower + sizeof(indx_t));
3213         mp->mp_ptrs[indx] = ofs;
3214         mp->mp_upper = ofs;
3215         mp->mp_lower += sizeof(indx_t);
3216
3217         /* Write the node data. */
3218         node = NODEPTR(mp, indx);
3219         node->mn_ksize = (key == NULL) ? 0 : key->mv_size;
3220         node->mn_flags = flags;
3221         if (IS_LEAF(mp))
3222                 SETDSZ(node,data->mv_size);
3223         else
3224                 SETPGNO(node,pgno);
3225
3226         if (key)
3227                 memcpy(NODEKEY(node), key->mv_data, key->mv_size);
3228
3229         if (IS_LEAF(mp)) {
3230                 assert(key);
3231                 if (ofp == NULL) {
3232                         if (F_ISSET(flags, F_BIGDATA))
3233                                 memcpy(node->mn_data + key->mv_size, data->mv_data,
3234                                     sizeof(pgno_t));
3235                         else
3236                                 memcpy(node->mn_data + key->mv_size, data->mv_data,
3237                                     data->mv_size);
3238                 } else {
3239                         memcpy(node->mn_data + key->mv_size, &ofp->mp_pgno,
3240                             sizeof(pgno_t));
3241                         memcpy(METADATA(ofp), data->mv_data, data->mv_size);
3242                 }
3243         }
3244
3245         return MDB_SUCCESS;
3246 }
3247
3248 static void
3249 mdb_del_node(MDB_page *mp, indx_t indx, int ksize)
3250 {
3251         unsigned int     sz;
3252         indx_t           i, j, numkeys, ptr;
3253         MDB_node        *node;
3254         char            *base;
3255
3256         DPRINTF("delete node %u on %s page %lu", indx,
3257             IS_LEAF(mp) ? "leaf" : "branch", mp->mp_pgno);
3258         assert(indx < NUMKEYS(mp));
3259
3260         if (IS_LEAF2(mp)) {
3261                 int x = NUMKEYS(mp) - 1 - indx;
3262                 base = LEAF2KEY(mp, indx, ksize);
3263                 if (x)
3264                         memmove(base, base + ksize, x * ksize);
3265                 mp->mp_lower -= sizeof(indx_t);
3266                 mp->mp_upper += ksize - sizeof(indx_t);
3267                 return;
3268         }
3269
3270         node = NODEPTR(mp, indx);
3271         sz = NODESIZE + node->mn_ksize;
3272         if (IS_LEAF(mp)) {
3273                 if (F_ISSET(node->mn_flags, F_BIGDATA))
3274                         sz += sizeof(pgno_t);
3275                 else
3276                         sz += NODEDSZ(node);
3277         }
3278         sz += sz & 1;
3279
3280         ptr = mp->mp_ptrs[indx];
3281         numkeys = NUMKEYS(mp);
3282         for (i = j = 0; i < numkeys; i++) {
3283                 if (i != indx) {
3284                         mp->mp_ptrs[j] = mp->mp_ptrs[i];
3285                         if (mp->mp_ptrs[i] < ptr)
3286                                 mp->mp_ptrs[j] += sz;
3287                         j++;
3288                 }
3289         }
3290
3291         base = (char *)mp + mp->mp_upper;
3292         memmove(base + sz, base, ptr - mp->mp_upper);
3293
3294         mp->mp_lower -= sizeof(indx_t);
3295         mp->mp_upper += sz;
3296 }
3297
3298 static void
3299 mdb_xcursor_init0(MDB_cursor *mc)
3300 {
3301         MDB_xcursor *mx = mc->mc_xcursor;
3302         MDB_dbi dbn;
3303
3304         mx->mx_txn = *mc->mc_txn;
3305         mx->mx_txn.mt_dbxs = mx->mx_dbxs;
3306         mx->mx_txn.mt_dbs = mx->mx_dbs;
3307         mx->mx_dbxs[0] = mc->mc_txn->mt_dbxs[0];
3308         mx->mx_dbxs[1] = mc->mc_txn->mt_dbxs[1];
3309         if (mc->mc_dbi > 1) {
3310                 mx->mx_dbxs[2] = mc->mc_txn->mt_dbxs[mc->mc_dbi];
3311                 dbn = 2;
3312         } else {
3313                 dbn = 1;
3314         }
3315         mx->mx_dbxs[dbn+1].md_parent = dbn;
3316         mx->mx_dbxs[dbn+1].md_cmp = mx->mx_dbxs[dbn].md_dcmp;
3317         mx->mx_dbxs[dbn+1].md_rel = mx->mx_dbxs[dbn].md_rel;
3318         mx->mx_dbxs[dbn+1].md_dirty = 0;
3319         mx->mx_txn.mt_numdbs = dbn+2;
3320         mx->mx_txn.mt_u = mc->mc_txn->mt_u;
3321
3322         mx->mx_cursor.mc_xcursor = NULL;
3323         mx->mx_cursor.mc_txn = &mx->mx_txn;
3324         mx->mx_cursor.mc_dbi = dbn+1;
3325 }
3326
3327 static void
3328 mdb_xcursor_init1(MDB_cursor *mc, MDB_node *node)
3329 {
3330         MDB_db *db = NODEDATA(node);
3331         MDB_xcursor *mx = mc->mc_xcursor;
3332         MDB_dbi dbn;
3333         mx->mx_dbs[0] = mc->mc_txn->mt_dbs[0];
3334         mx->mx_dbs[1] = mc->mc_txn->mt_dbs[1];
3335         if (mc->mc_dbi > 1) {
3336                 mx->mx_dbs[2] = mc->mc_txn->mt_dbs[mc->mc_dbi];
3337                 mx->mx_dbxs[2].md_dirty = mc->mc_txn->mt_dbxs[mc->mc_dbi].md_dirty;
3338                 dbn = 3;
3339         } else {
3340                 dbn = 2;
3341         }
3342         DPRINTF("Sub-db %u for db %u root page %lu", dbn, mc->mc_dbi, db->md_root);
3343         mx->mx_dbs[dbn] = *db;
3344         if (F_ISSET(mc->mc_pg[mc->mc_top]->mp_flags, P_DIRTY))
3345                 mx->mx_dbxs[dbn].md_dirty = 1;
3346         mx->mx_dbxs[dbn].md_name.mv_data = NODEKEY(node);
3347         mx->mx_dbxs[dbn].md_name.mv_size = node->mn_ksize;
3348         mx->mx_txn.mt_next_pgno = mc->mc_txn->mt_next_pgno;
3349         mx->mx_cursor.mc_snum = 0;
3350         mx->mx_cursor.mc_flags = 0;
3351 }
3352
3353 static void
3354 mdb_xcursor_init2(MDB_cursor *mc)
3355 {
3356         MDB_xcursor *mx = mc->mc_xcursor;
3357         MDB_dbi dbn;
3358         mx->mx_dbs[0] = mc->mc_txn->mt_dbs[0];
3359         mx->mx_dbs[1] = mc->mc_txn->mt_dbs[1];
3360         if (mc->mc_dbi > 1) {
3361                 mx->mx_dbs[2] = mc->mc_txn->mt_dbs[mc->mc_dbi];
3362                 mx->mx_dbxs[2].md_dirty = mc->mc_txn->mt_dbxs[mc->mc_dbi].md_dirty;
3363                 dbn = 3;
3364         } else {
3365                 dbn = 2;
3366         }
3367         DPRINTF("Sub-db %u for db %u root page %lu", dbn, mc->mc_dbi,
3368                 mx->mx_dbs[dbn].md_root);
3369         mx->mx_txn.mt_next_pgno = mc->mc_txn->mt_next_pgno;
3370 }
3371
3372 static void
3373 mdb_xcursor_fini(MDB_cursor *mc)
3374 {
3375         MDB_xcursor *mx = mc->mc_xcursor;
3376         mc->mc_txn->mt_next_pgno = mx->mx_txn.mt_next_pgno;
3377         mc->mc_txn->mt_dbs[0] = mx->mx_dbs[0];
3378         mc->mc_txn->mt_dbs[1] = mx->mx_dbs[1];
3379         if (mc->mc_dbi > 1) {
3380                 mc->mc_txn->mt_dbs[mc->mc_dbi] = mx->mx_dbs[2];
3381                 mc->mc_txn->mt_dbxs[mc->mc_dbi].md_dirty = mx->mx_dbxs[2].md_dirty;
3382         }
3383 }
3384
3385 int
3386 mdb_cursor_open(MDB_txn *txn, MDB_dbi dbi, MDB_cursor **ret)
3387 {
3388         MDB_cursor      *mc;
3389         size_t size = sizeof(MDB_cursor);
3390
3391         if (txn == NULL || ret == NULL || !dbi || dbi >= txn->mt_numdbs)
3392                 return EINVAL;
3393
3394         if (txn->mt_dbs[dbi].md_flags & MDB_DUPSORT)
3395                 size += sizeof(MDB_xcursor);
3396
3397         if ((mc = calloc(1, size)) != NULL) {
3398                 mc->mc_dbi = dbi;
3399                 mc->mc_txn = txn;
3400                 if (txn->mt_dbs[dbi].md_flags & MDB_DUPSORT) {
3401                         MDB_xcursor *mx = (MDB_xcursor *)(mc + 1);
3402                         mc->mc_xcursor = mx;
3403                         mdb_xcursor_init0(mc);
3404                 }
3405         } else {
3406                 return ENOMEM;
3407         }
3408
3409         *ret = mc;
3410
3411         return MDB_SUCCESS;
3412 }
3413
3414 /* Return the count of duplicate data items for the current key */
3415 int
3416 mdb_cursor_count(MDB_cursor *mc, unsigned long *countp)
3417 {
3418         MDB_node        *leaf;
3419
3420         if (mc == NULL || countp == NULL)
3421                 return EINVAL;
3422
3423         if (!(mc->mc_txn->mt_dbs[mc->mc_dbi].md_flags & MDB_DUPSORT))
3424                 return EINVAL;
3425
3426         leaf = NODEPTR(mc->mc_pg[mc->mc_top], mc->mc_ki[mc->mc_top]);
3427         if (!F_ISSET(leaf->mn_flags, F_DUPDATA)) {
3428                 *countp = 1;
3429         } else {
3430                 if (!(mc->mc_xcursor->mx_cursor.mc_flags & C_INITIALIZED))
3431                         return EINVAL;
3432
3433                 *countp = mc->mc_xcursor->mx_txn.mt_dbs[mc->mc_xcursor->mx_cursor.mc_dbi].md_entries;
3434         }
3435         return MDB_SUCCESS;
3436 }
3437
3438 void
3439 mdb_cursor_close(MDB_cursor *mc)
3440 {
3441         if (mc != NULL) {
3442                 free(mc);
3443         }
3444 }
3445
3446 static int
3447 mdb_update_key(MDB_page *mp, indx_t indx, MDB_val *key)
3448 {
3449         indx_t                   ptr, i, numkeys;
3450         int                      delta;
3451         size_t                   len;
3452         MDB_node                *node;
3453         char                    *base;
3454         DKBUF;
3455
3456         node = NODEPTR(mp, indx);
3457         ptr = mp->mp_ptrs[indx];
3458         DPRINTF("update key %u (ofs %u) [%.*s] to [%s] on page %lu",
3459             indx, ptr,
3460             (int)node->mn_ksize, (char *)NODEKEY(node),
3461                 DKEY(key),
3462             mp->mp_pgno);
3463
3464         delta = key->mv_size - node->mn_ksize;
3465         if (delta) {
3466                 if (delta > 0 && SIZELEFT(mp) < delta) {
3467                         DPRINTF("OUCH! Not enough room, delta = %d", delta);
3468                         return ENOSPC;
3469                 }
3470
3471                 numkeys = NUMKEYS(mp);
3472                 for (i = 0; i < numkeys; i++) {
3473                         if (mp->mp_ptrs[i] <= ptr)
3474                                 mp->mp_ptrs[i] -= delta;
3475                 }
3476
3477                 base = (char *)mp + mp->mp_upper;
3478                 len = ptr - mp->mp_upper + NODESIZE;
3479                 memmove(base - delta, base, len);
3480                 mp->mp_upper -= delta;
3481
3482                 node = NODEPTR(mp, indx);
3483                 node->mn_ksize = key->mv_size;
3484         }
3485
3486         memcpy(NODEKEY(node), key->mv_data, key->mv_size);
3487
3488         return MDB_SUCCESS;
3489 }
3490
3491 /* Move a node from csrc to cdst.
3492  */
3493 static int
3494 mdb_move_node(MDB_cursor *csrc, MDB_cursor *cdst)
3495 {
3496         int                      rc;
3497         MDB_node                *srcnode;
3498         MDB_val          key, data;
3499         DKBUF;
3500
3501         /* Mark src and dst as dirty. */
3502         if ((rc = mdb_touch(csrc)) ||
3503             (rc = mdb_touch(cdst)))
3504                 return rc;;
3505
3506         if (IS_LEAF2(csrc->mc_pg[csrc->mc_top])) {
3507                 srcnode = NODEPTR(csrc->mc_pg[csrc->mc_top], 0);        /* fake */
3508                 key.mv_size = csrc->mc_txn->mt_dbs[csrc->mc_dbi].md_pad;
3509                 key.mv_data = LEAF2KEY(csrc->mc_pg[csrc->mc_top], csrc->mc_ki[csrc->mc_top], key.mv_size);
3510                 data.mv_size = 0;
3511                 data.mv_data = NULL;
3512         } else {
3513                 if (csrc->mc_ki[csrc->mc_top] == 0 && IS_BRANCH(csrc->mc_pg[csrc->mc_top])) {
3514                         unsigned int snum = csrc->mc_snum;
3515                         /* must find the lowest key below src */
3516                         mdb_search_page_root(csrc, NULL, 0);
3517                         srcnode = NODEPTR(csrc->mc_pg[csrc->mc_top], 0);
3518                         csrc->mc_snum = snum--;
3519                         csrc->mc_top = snum;
3520                 } else {
3521                         srcnode = NODEPTR(csrc->mc_pg[csrc->mc_top], csrc->mc_ki[csrc->mc_top]);
3522                 }
3523                 key.mv_size = NODEKSZ(srcnode);
3524                 key.mv_data = NODEKEY(srcnode);
3525                 data.mv_size = NODEDSZ(srcnode);
3526                 data.mv_data = NODEDATA(srcnode);
3527         }
3528         DPRINTF("moving %s node %u [%s] on page %lu to node %u on page %lu",
3529             IS_LEAF(csrc->mc_pg[csrc->mc_top]) ? "leaf" : "branch",
3530             csrc->mc_ki[csrc->mc_top],
3531                 DKEY(&key),
3532             csrc->mc_pg[csrc->mc_top]->mp_pgno,
3533             cdst->mc_ki[cdst->mc_top], cdst->mc_pg[cdst->mc_top]->mp_pgno);
3534
3535         /* Add the node to the destination page.
3536          */
3537         rc = mdb_add_node(cdst, cdst->mc_ki[cdst->mc_top], &key, &data, NODEPGNO(srcnode),
3538             srcnode->mn_flags);
3539         if (rc != MDB_SUCCESS)
3540                 return rc;
3541
3542         /* Delete the node from the source page.
3543          */
3544         mdb_del_node(csrc->mc_pg[csrc->mc_top], csrc->mc_ki[csrc->mc_top], key.mv_size);
3545
3546         /* Update the parent separators.
3547          */
3548         if (csrc->mc_ki[csrc->mc_top] == 0) {
3549                 if (csrc->mc_ki[csrc->mc_top-1] != 0) {
3550                         if (IS_LEAF2(csrc->mc_pg[csrc->mc_top])) {
3551                                 key.mv_data = LEAF2KEY(csrc->mc_pg[csrc->mc_top], csrc->mc_ki[csrc->mc_top], key.mv_size);
3552                         } else {
3553                                 srcnode = NODEPTR(csrc->mc_pg[csrc->mc_top], csrc->mc_ki[csrc->mc_top]);
3554                                 key.mv_size = NODEKSZ(srcnode);
3555                                 key.mv_data = NODEKEY(srcnode);
3556                         }
3557                         DPRINTF("update separator for source page %lu to [%s]",
3558                                 csrc->mc_pg[csrc->mc_top]->mp_pgno, DKEY(&key));
3559                         if ((rc = mdb_update_key(csrc->mc_pg[csrc->mc_top-1], csrc->mc_ki[csrc->mc_top-1],
3560                                 &key)) != MDB_SUCCESS)
3561                                 return rc;
3562                 }
3563                 if (IS_BRANCH(csrc->mc_pg[csrc->mc_top])) {
3564                         MDB_val  nullkey;
3565                         nullkey.mv_size = 0;
3566                         assert(mdb_update_key(csrc->mc_pg[csrc->mc_top], 0, &nullkey) == MDB_SUCCESS);
3567                 }
3568         }
3569
3570         if (cdst->mc_ki[cdst->mc_top] == 0) {
3571                 if (cdst->mc_ki[cdst->mc_top-1] != 0) {
3572                         if (IS_LEAF2(csrc->mc_pg[csrc->mc_top])) {
3573                                 key.mv_data = LEAF2KEY(cdst->mc_pg[cdst->mc_top], 0, key.mv_size);
3574                         } else {
3575                                 srcnode = NODEPTR(cdst->mc_pg[cdst->mc_top], 0);
3576                                 key.mv_size = NODEKSZ(srcnode);
3577                                 key.mv_data = NODEKEY(srcnode);
3578                         }
3579                         DPRINTF("update separator for destination page %lu to [%s]",
3580                                 cdst->mc_pg[cdst->mc_top]->mp_pgno, DKEY(&key));
3581                         if ((rc = mdb_update_key(cdst->mc_pg[cdst->mc_top-1], cdst->mc_ki[cdst->mc_top-1],
3582                                 &key)) != MDB_SUCCESS)
3583                                 return rc;
3584                 }
3585                 if (IS_BRANCH(cdst->mc_pg[cdst->mc_top])) {
3586                         MDB_val  nullkey;
3587                         nullkey.mv_size = 0;
3588                         assert(mdb_update_key(cdst->mc_pg[cdst->mc_top], 0, &nullkey) == MDB_SUCCESS);
3589                 }
3590         }
3591
3592         return MDB_SUCCESS;
3593 }
3594
3595 static int
3596 mdb_merge(MDB_cursor *csrc, MDB_cursor *cdst)
3597 {
3598         int                      rc;
3599         indx_t                   i, j;
3600         MDB_node                *srcnode;
3601         MDB_val          key, data;
3602
3603         DPRINTF("merging page %lu into %lu", csrc->mc_pg[csrc->mc_top]->mp_pgno, cdst->mc_pg[cdst->mc_top]->mp_pgno);
3604
3605         assert(csrc->mc_snum > 1);      /* can't merge root page */
3606         assert(cdst->mc_snum > 1);
3607
3608         /* Mark dst as dirty. */
3609         if ((rc = mdb_touch(cdst)))
3610                 return rc;
3611
3612         /* Move all nodes from src to dst.
3613          */
3614         j = NUMKEYS(cdst->mc_pg[cdst->mc_top]);
3615         if (IS_LEAF2(csrc->mc_pg[csrc->mc_top])) {
3616                 key.mv_size = csrc->mc_txn->mt_dbs[csrc->mc_dbi].md_pad;
3617                 key.mv_data = METADATA(csrc->mc_pg[csrc->mc_top]);
3618                 for (i = 0; i < NUMKEYS(csrc->mc_pg[csrc->mc_top]); i++, j++) {
3619                         rc = mdb_add_node(cdst, j, &key, NULL, 0, 0);
3620                         if (rc != MDB_SUCCESS)
3621                                 return rc;
3622                         key.mv_data = (char *)key.mv_data + key.mv_size;
3623                 }
3624         } else {
3625                 for (i = 0; i < NUMKEYS(csrc->mc_pg[csrc->mc_top]); i++, j++) {
3626                         srcnode = NODEPTR(csrc->mc_pg[csrc->mc_top], i);
3627
3628                         key.mv_size = srcnode->mn_ksize;
3629                         key.mv_data = NODEKEY(srcnode);
3630                         data.mv_size = NODEDSZ(srcnode);
3631                         data.mv_data = NODEDATA(srcnode);
3632                         rc = mdb_add_node(cdst, j, &key, &data, NODEPGNO(srcnode), srcnode->mn_flags);
3633                         if (rc != MDB_SUCCESS)
3634                                 return rc;
3635                 }
3636         }
3637
3638         DPRINTF("dst page %lu now has %u keys (%.1f%% filled)",
3639             cdst->mc_pg[cdst->mc_top]->mp_pgno, NUMKEYS(cdst->mc_pg[cdst->mc_top]), (float)PAGEFILL(cdst->mc_txn->mt_env, cdst->mc_pg[cdst->mc_top]) / 10);
3640
3641         /* Unlink the src page from parent and add to free list.
3642          */
3643         mdb_del_node(csrc->mc_pg[csrc->mc_top-1], csrc->mc_ki[csrc->mc_top-1], 0);
3644         if (csrc->mc_ki[csrc->mc_top-1] == 0) {
3645                 key.mv_size = 0;
3646                 if ((rc = mdb_update_key(csrc->mc_pg[csrc->mc_top-1], 0, &key)) != MDB_SUCCESS)
3647                         return rc;
3648         }
3649
3650         mdb_midl_append(csrc->mc_txn->mt_free_pgs, csrc->mc_pg[csrc->mc_top]->mp_pgno);
3651         if (IS_LEAF(csrc->mc_pg[csrc->mc_top]))
3652                 csrc->mc_txn->mt_dbs[csrc->mc_dbi].md_leaf_pages--;
3653         else
3654                 csrc->mc_txn->mt_dbs[csrc->mc_dbi].md_branch_pages--;
3655         cursor_pop_page(csrc);
3656
3657         return mdb_rebalance(csrc);
3658 }
3659
3660 static void
3661 mdb_cursor_copy(const MDB_cursor *csrc, MDB_cursor *cdst)
3662 {
3663         unsigned int i;
3664
3665         cdst->mc_txn = csrc->mc_txn;
3666         cdst->mc_dbi = csrc->mc_dbi;
3667         cdst->mc_snum = csrc->mc_snum;
3668         cdst->mc_top = csrc->mc_top;
3669         cdst->mc_flags = csrc->mc_flags;
3670
3671         for (i=0; i<csrc->mc_snum; i++) {
3672                 cdst->mc_pg[i] = csrc->mc_pg[i];
3673                 cdst->mc_ki[i] = csrc->mc_ki[i];
3674         }
3675 }
3676
3677 #define FILL_THRESHOLD   250
3678
3679 static int
3680 mdb_rebalance(MDB_cursor *mc)
3681 {
3682         MDB_node        *node;
3683         MDB_page        *root;
3684         int rc;
3685         unsigned int ptop;
3686         MDB_cursor      mn;
3687
3688         DPRINTF("rebalancing %s page %lu (has %u keys, %.1f%% full)",
3689             IS_LEAF(mc->mc_pg[mc->mc_top]) ? "leaf" : "branch",
3690             mc->mc_pg[mc->mc_top]->mp_pgno, NUMKEYS(mc->mc_pg[mc->mc_top]), (float)PAGEFILL(mc->mc_txn->mt_env, mc->mc_pg[mc->mc_top]) / 10);
3691
3692         if (PAGEFILL(mc->mc_txn->mt_env, mc->mc_pg[mc->mc_top]) >= FILL_THRESHOLD) {
3693                 DPRINTF("no need to rebalance page %lu, above fill threshold",
3694                     mc->mc_pg[mc->mc_top]->mp_pgno);
3695                 return MDB_SUCCESS;
3696         }
3697
3698         if (mc->mc_snum < 2) {
3699                 if (NUMKEYS(mc->mc_pg[mc->mc_top]) == 0) {
3700                         DPUTS("tree is completely empty");
3701                         mc->mc_txn->mt_dbs[mc->mc_dbi].md_root = P_INVALID;
3702                         mc->mc_txn->mt_dbs[mc->mc_dbi].md_depth = 0;
3703                         mc->mc_txn->mt_dbs[mc->mc_dbi].md_leaf_pages = 0;
3704                         mdb_midl_append(mc->mc_txn->mt_free_pgs, mc->mc_pg[mc->mc_top]->mp_pgno);
3705                 } else if (IS_BRANCH(mc->mc_pg[mc->mc_top]) && NUMKEYS(mc->mc_pg[mc->mc_top]) == 1) {
3706                         DPUTS("collapsing root page!");
3707                         mdb_midl_append(mc->mc_txn->mt_free_pgs, mc->mc_pg[mc->mc_top]->mp_pgno);
3708                         mc->mc_txn->mt_dbs[mc->mc_dbi].md_root = NODEPGNO(NODEPTR(mc->mc_pg[mc->mc_top], 0));
3709                         if ((rc = mdb_get_page(mc->mc_txn, mc->mc_txn->mt_dbs[mc->mc_dbi].md_root, &root)))
3710                                 return rc;
3711                         mc->mc_txn->mt_dbs[mc->mc_dbi].md_depth--;
3712                         mc->mc_txn->mt_dbs[mc->mc_dbi].md_branch_pages--;
3713                 } else
3714                         DPUTS("root page doesn't need rebalancing");
3715                 return MDB_SUCCESS;
3716         }
3717
3718         /* The parent (branch page) must have at least 2 pointers,
3719          * otherwise the tree is invalid.
3720          */
3721         ptop = mc->mc_top-1;
3722         assert(NUMKEYS(mc->mc_pg[ptop]) > 1);
3723
3724         /* Leaf page fill factor is below the threshold.
3725          * Try to move keys from left or right neighbor, or
3726          * merge with a neighbor page.
3727          */
3728
3729         /* Find neighbors.
3730          */
3731         mdb_cursor_copy(mc, &mn);
3732         mn.mc_xcursor = NULL;
3733
3734         if (mc->mc_ki[ptop] == 0) {
3735                 /* We're the leftmost leaf in our parent.
3736                  */
3737                 DPUTS("reading right neighbor");
3738                 mn.mc_ki[ptop]++;
3739                 node = NODEPTR(mc->mc_pg[ptop], mn.mc_ki[ptop]);
3740                 if ((rc = mdb_get_page(mc->mc_txn, NODEPGNO(node), &mn.mc_pg[mn.mc_top])))
3741                         return rc;
3742                 mn.mc_ki[mn.mc_top] = 0;
3743                 mc->mc_ki[mc->mc_top] = NUMKEYS(mc->mc_pg[mc->mc_top]);
3744         } else {
3745                 /* There is at least one neighbor to the left.
3746                  */
3747                 DPUTS("reading left neighbor");
3748                 mn.mc_ki[ptop]--;
3749                 node = NODEPTR(mc->mc_pg[ptop], mn.mc_ki[ptop]);
3750                 if ((rc = mdb_get_page(mc->mc_txn, NODEPGNO(node), &mn.mc_pg[mn.mc_top])))
3751                         return rc;
3752                 mn.mc_ki[mn.mc_top] = NUMKEYS(mn.mc_pg[mn.mc_top]) - 1;
3753                 mc->mc_ki[mc->mc_top] = 0;
3754         }
3755
3756         DPRINTF("found neighbor page %lu (%u keys, %.1f%% full)",
3757             mn.mc_pg[mn.mc_top]->mp_pgno, NUMKEYS(mn.mc_pg[mn.mc_top]), (float)PAGEFILL(mc->mc_txn->mt_env, mn.mc_pg[mn.mc_top]) / 10);
3758
3759         /* If the neighbor page is above threshold and has at least two
3760          * keys, move one key from it.
3761          *
3762          * Otherwise we should try to merge them.
3763          */
3764         if (PAGEFILL(mc->mc_txn->mt_env, mn.mc_pg[mn.mc_top]) >= FILL_THRESHOLD && NUMKEYS(mn.mc_pg[mn.mc_top]) >= 2)
3765                 return mdb_move_node(&mn, mc);
3766         else { /* FIXME: if (has_enough_room()) */
3767                 if (mc->mc_ki[ptop] == 0)
3768                         return mdb_merge(&mn, mc);
3769                 else
3770                         return mdb_merge(mc, &mn);
3771         }
3772 }
3773
3774 static int
3775 mdb_del0(MDB_cursor *mc, MDB_node *leaf)
3776 {
3777         int rc;
3778
3779         /* add overflow pages to free list */
3780         if (!IS_LEAF2(mc->mc_pg[mc->mc_top]) && F_ISSET(leaf->mn_flags, F_BIGDATA)) {
3781                 int i, ovpages;
3782                 pgno_t pg;
3783
3784                 memcpy(&pg, NODEDATA(leaf), sizeof(pg));
3785                 ovpages = OVPAGES(NODEDSZ(leaf), mc->mc_txn->mt_env->me_psize);
3786                 for (i=0; i<ovpages; i++) {
3787                         DPRINTF("freed ov page %lu", pg);
3788                         mdb_midl_append(mc->mc_txn->mt_free_pgs, pg);
3789                         pg++;
3790                 }
3791         }
3792         mdb_del_node(mc->mc_pg[mc->mc_top], mc->mc_ki[mc->mc_top], mc->mc_txn->mt_dbs[mc->mc_dbi].md_pad);
3793         mc->mc_txn->mt_dbs[mc->mc_dbi].md_entries--;
3794         rc = mdb_rebalance(mc);
3795         if (rc != MDB_SUCCESS)
3796                 mc->mc_txn->mt_flags |= MDB_TXN_ERROR;
3797
3798         return rc;
3799 }
3800
3801 int
3802 mdb_del(MDB_txn *txn, MDB_dbi dbi,
3803     MDB_val *key, MDB_val *data)
3804 {
3805         MDB_cursor mc;
3806         MDB_xcursor mx;
3807         MDB_cursor_op op;
3808         MDB_val rdata, *xdata;
3809         int              rc, exact;
3810         DKBUF;
3811
3812         assert(key != NULL);
3813
3814         DPRINTF("====> delete db %u key [%s]", dbi, DKEY(key));
3815
3816         if (txn == NULL || !dbi || dbi >= txn->mt_numdbs)
3817                 return EINVAL;
3818
3819         if (F_ISSET(txn->mt_flags, MDB_TXN_RDONLY)) {
3820                 return EACCES;
3821         }
3822
3823         if (key->mv_size == 0 || key->mv_size > MAXKEYSIZE) {
3824                 return EINVAL;
3825         }
3826
3827         mc.mc_txn = txn;
3828         mc.mc_dbi = dbi;
3829         mc.mc_flags = 0;
3830         if (txn->mt_dbs[dbi].md_flags & MDB_DUPSORT) {
3831                 mc.mc_xcursor = &mx;
3832                 mdb_xcursor_init0(&mc);
3833         } else {
3834                 mc.mc_xcursor = NULL;
3835         }
3836
3837         exact = 0;
3838         if (data) {
3839                 op = MDB_GET_BOTH;
3840                 rdata = *data;
3841                 xdata = &rdata;
3842         } else {
3843                 op = MDB_SET;
3844                 xdata = NULL;
3845         }
3846         rc = mdb_cursor_set(&mc, key, xdata, op, &exact);
3847         if (rc == 0)
3848                 rc = mdb_cursor_del(&mc, data ? 0 : MDB_NODUPDATA);
3849         return rc;
3850 }
3851
3852 /* Split page <mc->top>, and insert <key,(data|newpgno)> in either left or
3853  * right sibling, at index <mc->ki> (as if unsplit). Updates mc->top and
3854  * mc->ki with the actual values after split, ie if mc->top and mc->ki
3855  * refer to a node in the new right sibling page.
3856  */
3857 static int
3858 mdb_split(MDB_cursor *mc, MDB_val *newkey, MDB_val *newdata, pgno_t newpgno)
3859 {
3860         uint8_t          flags;
3861         int              rc = MDB_SUCCESS, ins_new = 0;
3862         indx_t           newindx;
3863         pgno_t           pgno = 0;
3864         unsigned int     i, j, split_indx, nkeys, pmax;
3865         MDB_node        *node;
3866         MDB_val  sepkey, rkey, rdata;
3867         MDB_page        *copy;
3868         MDB_page        *mp, *rp, *pp;
3869         unsigned int ptop;
3870         MDB_cursor      mn;
3871         DKBUF;
3872
3873         mp = mc->mc_pg[mc->mc_top];
3874         newindx = mc->mc_ki[mc->mc_top];
3875
3876         DPRINTF("-----> splitting %s page %lu and adding [%s] at index %i",
3877             IS_LEAF(mp) ? "leaf" : "branch", mp->mp_pgno,
3878             DKEY(newkey), mc->mc_ki[mc->mc_top]);
3879
3880         if (mc->mc_snum < 2) {
3881                 if ((pp = mdb_new_page(mc, P_BRANCH, 1)) == NULL)
3882                         return ENOMEM;
3883                 /* shift current top to make room for new parent */
3884                 mc->mc_pg[1] = mc->mc_pg[0];
3885                 mc->mc_ki[1] = mc->mc_ki[0];
3886                 mc->mc_pg[0] = pp;
3887                 mc->mc_ki[0] = 0;
3888                 mc->mc_txn->mt_dbs[mc->mc_dbi].md_root = pp->mp_pgno;
3889                 DPRINTF("root split! new root = %lu", pp->mp_pgno);
3890                 mc->mc_txn->mt_dbs[mc->mc_dbi].md_depth++;
3891
3892                 /* Add left (implicit) pointer. */
3893                 if ((rc = mdb_add_node(mc, 0, NULL, NULL, mp->mp_pgno, 0)) != MDB_SUCCESS) {
3894                         /* undo the pre-push */
3895                         mc->mc_pg[0] = mc->mc_pg[1];
3896                         mc->mc_ki[0] = mc->mc_ki[1];
3897                         mc->mc_txn->mt_dbs[mc->mc_dbi].md_root = mp->mp_pgno;
3898                         mc->mc_txn->mt_dbs[mc->mc_dbi].md_depth--;
3899                         return rc;
3900                 }
3901                 mc->mc_snum = 2;
3902                 mc->mc_top = 1;
3903                 ptop = 0;
3904         } else {
3905                 ptop = mc->mc_top-1;
3906                 DPRINTF("parent branch page is %lu", mc->mc_pg[ptop]->mp_pgno);
3907         }
3908
3909         /* Create a right sibling. */
3910         if ((rp = mdb_new_page(mc, mp->mp_flags, 1)) == NULL)
3911                 return ENOMEM;
3912         mdb_cursor_copy(mc, &mn);
3913         mn.mc_pg[mn.mc_top] = rp;
3914         mn.mc_ki[ptop] = mc->mc_ki[ptop]+1;
3915         DPRINTF("new right sibling: page %lu", rp->mp_pgno);
3916
3917         nkeys = NUMKEYS(mp);
3918         split_indx = nkeys / 2 + 1;
3919
3920         if (IS_LEAF2(rp)) {
3921                 char *split, *ins;
3922                 int x;
3923                 unsigned int lsize, rsize, ksize;
3924                 /* Move half of the keys to the right sibling */
3925                 copy = NULL;
3926                 x = mc->mc_ki[mc->mc_top] - split_indx;
3927                 ksize = mc->mc_txn->mt_dbs[mc->mc_dbi].md_pad;
3928                 split = LEAF2KEY(mp, split_indx, ksize);
3929                 rsize = (nkeys - split_indx) * ksize;
3930                 lsize = (nkeys - split_indx) * sizeof(indx_t);
3931                 mp->mp_lower -= lsize;
3932                 rp->mp_lower += lsize;
3933                 mp->mp_upper += rsize - lsize;
3934                 rp->mp_upper -= rsize - lsize;
3935                 sepkey.mv_size = ksize;
3936                 if (newindx == split_indx) {
3937                         sepkey.mv_data = newkey->mv_data;
3938                 } else {
3939                         sepkey.mv_data = split;
3940                 }
3941                 if (x<0) {
3942                         ins = LEAF2KEY(mp, mc->mc_ki[mc->mc_top], ksize);
3943                         memcpy(rp->mp_ptrs, split, rsize);
3944                         sepkey.mv_data = rp->mp_ptrs;
3945                         memmove(ins+ksize, ins, (split_indx - mc->mc_ki[mc->mc_top]) * ksize);
3946                         memcpy(ins, newkey->mv_data, ksize);
3947                         mp->mp_lower += sizeof(indx_t);
3948                         mp->mp_upper -= ksize - sizeof(indx_t);
3949                 } else {
3950                         if (x)
3951                                 memcpy(rp->mp_ptrs, split, x * ksize);
3952                         ins = LEAF2KEY(rp, x, ksize);
3953                         memcpy(ins, newkey->mv_data, ksize);
3954                         memcpy(ins+ksize, split + x * ksize, rsize - x * ksize);
3955                         rp->mp_lower += sizeof(indx_t);
3956                         rp->mp_upper -= ksize - sizeof(indx_t);
3957                         mc->mc_ki[mc->mc_top] = x;
3958                         mc->mc_pg[mc->mc_top] = rp;
3959                 }
3960                 goto newsep;
3961         }
3962
3963         /* For leaf pages, check the split point based on what
3964          * fits where, since otherwise add_node can fail.
3965          */
3966         if (IS_LEAF(mp)) {
3967                 unsigned int psize, nsize;
3968                 /* Maximum free space in an empty page */
3969                 pmax = mc->mc_txn->mt_env->me_psize - PAGEHDRSZ;
3970                 nsize = mdb_leaf_size(mc->mc_txn->mt_env, newkey, newdata);
3971                 if (newindx < split_indx) {
3972                         psize = nsize;
3973                         for (i=0; i<split_indx; i++) {
3974                                 node = NODEPTR(mp, i);
3975                                 psize += NODESIZE + NODEKSZ(node) + sizeof(indx_t);
3976                                 if (F_ISSET(node->mn_flags, F_BIGDATA))
3977                                         psize += sizeof(pgno_t);
3978                                 else
3979                                         psize += NODEDSZ(node);
3980                                 psize += psize & 1;
3981                                 if (psize > pmax) {
3982                                         split_indx = i;
3983                                         break;
3984                                 }
3985                         }
3986                 } else {
3987                         psize = nsize;
3988                         for (i=nkeys-1; i>=split_indx; i--) {
3989                                 node = NODEPTR(mp, i);
3990                                 psize += NODESIZE + NODEKSZ(node) + sizeof(indx_t);
3991                                 if (F_ISSET(node->mn_flags, F_BIGDATA))
3992                                         psize += sizeof(pgno_t);
3993                                 else
3994                                         psize += NODEDSZ(node);
3995                                 psize += psize & 1;
3996                                 if (psize > pmax) {
3997                                         split_indx = i+1;
3998                                         break;
3999                                 }
4000                         }
4001                 }
4002         }
4003
4004         /* First find the separating key between the split pages.
4005          */
4006         if (newindx == split_indx) {
4007                 sepkey.mv_size = newkey->mv_size;
4008                 sepkey.mv_data = newkey->mv_data;
4009         } else {
4010                 node = NODEPTR(mp, split_indx);
4011                 sepkey.mv_size = node->mn_ksize;
4012                 sepkey.mv_data = NODEKEY(node);
4013         }
4014
4015 newsep:
4016         DPRINTF("separator is [%s]", DKEY(&sepkey));
4017
4018         /* Copy separator key to the parent.
4019          */
4020         if (SIZELEFT(mn.mc_pg[ptop]) < mdb_branch_size(mc->mc_txn->mt_env, &sepkey)) {
4021                 mn.mc_snum--;
4022                 mn.mc_top--;
4023                 rc = mdb_split(&mn, &sepkey, NULL, rp->mp_pgno);
4024
4025                 /* Right page might now have changed parent.
4026                  * Check if left page also changed parent.
4027                  */
4028                 if (mn.mc_pg[ptop] != mc->mc_pg[ptop] &&
4029                     mc->mc_ki[ptop] >= NUMKEYS(mc->mc_pg[ptop])) {
4030                         mc->mc_pg[ptop] = mn.mc_pg[ptop];
4031                         mc->mc_ki[ptop] = mn.mc_ki[ptop] - 1;
4032                 }
4033         } else {
4034                 mn.mc_top--;
4035                 rc = mdb_add_node(&mn, mn.mc_ki[ptop], &sepkey, NULL, rp->mp_pgno, 0);
4036                 mn.mc_top++;
4037         }
4038         if (IS_LEAF2(rp)) {
4039                 return rc;
4040         }
4041         if (rc != MDB_SUCCESS) {
4042                 return rc;
4043         }
4044
4045         /* Move half of the keys to the right sibling. */
4046
4047         /* grab a page to hold a temporary copy */
4048         if (mc->mc_txn->mt_env->me_dpages) {
4049                 copy = mc->mc_txn->mt_env->me_dpages;
4050                 mc->mc_txn->mt_env->me_dpages = copy->mp_next;
4051         } else {
4052                 if ((copy = malloc(mc->mc_txn->mt_env->me_psize)) == NULL)
4053                         return ENOMEM;
4054         }
4055
4056         copy->mp_pgno  = mp->mp_pgno;
4057         copy->mp_flags = mp->mp_flags;
4058         copy->mp_lower = PAGEHDRSZ;
4059         copy->mp_upper = mc->mc_txn->mt_env->me_psize;
4060         mc->mc_pg[mc->mc_top] = copy;
4061         for (i = j = 0; i <= nkeys; j++) {
4062                 if (i == split_indx) {
4063                 /* Insert in right sibling. */
4064                 /* Reset insert index for right sibling. */
4065                         j = (i == newindx && ins_new);
4066                         mc->mc_pg[mc->mc_top] = rp;
4067                 }
4068
4069                 if (i == newindx && !ins_new) {
4070                         /* Insert the original entry that caused the split. */
4071                         rkey.mv_data = newkey->mv_data;
4072                         rkey.mv_size = newkey->mv_size;
4073                         if (IS_LEAF(mp)) {
4074                                 rdata.mv_data = newdata->mv_data;
4075                                 rdata.mv_size = newdata->mv_size;
4076                         } else
4077                                 pgno = newpgno;
4078                         flags = 0;
4079
4080                         ins_new = 1;
4081
4082                         /* Update page and index for the new key. */
4083                         mc->mc_ki[mc->mc_top] = j;
4084                 } else if (i == nkeys) {
4085                         break;
4086                 } else {
4087                         node = NODEPTR(mp, i);
4088                         rkey.mv_data = NODEKEY(node);
4089                         rkey.mv_size = node->mn_ksize;
4090                         if (IS_LEAF(mp)) {
4091                                 rdata.mv_data = NODEDATA(node);
4092                                 rdata.mv_size = NODEDSZ(node);
4093                         } else
4094                                 pgno = NODEPGNO(node);
4095                         flags = node->mn_flags;
4096
4097                         i++;
4098                 }
4099
4100                 if (!IS_LEAF(mp) && j == 0) {
4101                         /* First branch index doesn't need key data. */
4102                         rkey.mv_size = 0;
4103                 }
4104
4105                 rc = mdb_add_node(mc, j, &rkey, &rdata, pgno, flags);
4106         }
4107
4108         if (mc->mc_pg[mc->mc_top] == copy)
4109                 mc->mc_pg[mc->mc_top] = mp;
4110
4111         nkeys = NUMKEYS(copy);
4112         for (i=0; i<nkeys; i++)
4113                 mp->mp_ptrs[i] = copy->mp_ptrs[i];
4114         mp->mp_lower = copy->mp_lower;
4115         mp->mp_upper = copy->mp_upper;
4116         memcpy(NODEPTR(mp, nkeys-1), NODEPTR(copy, nkeys-1),
4117                 mc->mc_txn->mt_env->me_psize - copy->mp_upper);
4118
4119         /* return tmp page to freelist */
4120         copy->mp_next = mc->mc_txn->mt_env->me_dpages;
4121         mc->mc_txn->mt_env->me_dpages = copy;
4122         return rc;
4123 }
4124
4125 int
4126 mdb_put(MDB_txn *txn, MDB_dbi dbi,
4127     MDB_val *key, MDB_val *data, unsigned int flags)
4128 {
4129         MDB_cursor mc;
4130         MDB_xcursor mx;
4131
4132         assert(key != NULL);
4133         assert(data != NULL);
4134
4135         if (txn == NULL || !dbi || dbi >= txn->mt_numdbs)
4136                 return EINVAL;
4137
4138         if (F_ISSET(txn->mt_flags, MDB_TXN_RDONLY)) {
4139                 return EACCES;
4140         }
4141
4142         if (key->mv_size == 0 || key->mv_size > MAXKEYSIZE) {
4143                 return EINVAL;
4144         }
4145
4146         if ((flags & (MDB_NOOVERWRITE|MDB_NODUPDATA)) != flags)
4147                 return EINVAL;
4148
4149         mc.mc_txn = txn;
4150         mc.mc_dbi = dbi;
4151         mc.mc_snum = 0;
4152         mc.mc_flags = 0;
4153         if (txn->mt_dbs[dbi].md_flags & MDB_DUPSORT) {
4154                 mc.mc_xcursor = &mx;
4155                 mdb_xcursor_init0(&mc);
4156         } else {
4157                 mc.mc_xcursor = NULL;
4158         }
4159         return mdb_cursor_put(&mc, key, data, flags);
4160 }
4161
4162 int
4163 mdb_env_set_flags(MDB_env *env, unsigned int flag, int onoff)
4164 {
4165 #define CHANGEABLE      (MDB_NOSYNC)
4166         if ((flag & CHANGEABLE) != flag)
4167                 return EINVAL;
4168         if (onoff)
4169                 env->me_flags |= flag;
4170         else
4171                 env->me_flags &= ~flag;
4172         return MDB_SUCCESS;
4173 }
4174
4175 int
4176 mdb_env_get_flags(MDB_env *env, unsigned int *arg)
4177 {
4178         if (!env || !arg)
4179                 return EINVAL;
4180
4181         *arg = env->me_flags;
4182         return MDB_SUCCESS;
4183 }
4184
4185 int
4186 mdb_env_get_path(MDB_env *env, const char **arg)
4187 {
4188         if (!env || !arg)
4189                 return EINVAL;
4190
4191         *arg = env->me_path;
4192         return MDB_SUCCESS;
4193 }
4194
4195 static int
4196 mdb_stat0(MDB_env *env, MDB_db *db, MDB_stat *arg)
4197 {
4198         arg->ms_psize = env->me_psize;
4199         arg->ms_depth = db->md_depth;
4200         arg->ms_branch_pages = db->md_branch_pages;
4201         arg->ms_leaf_pages = db->md_leaf_pages;
4202         arg->ms_overflow_pages = db->md_overflow_pages;
4203         arg->ms_entries = db->md_entries;
4204
4205         return MDB_SUCCESS;
4206 }
4207 int
4208 mdb_env_stat(MDB_env *env, MDB_stat *arg)
4209 {
4210         int toggle;
4211
4212         if (env == NULL || arg == NULL)
4213                 return EINVAL;
4214
4215         mdb_env_read_meta(env, &toggle);
4216
4217         return mdb_stat0(env, &env->me_metas[toggle]->mm_dbs[MAIN_DBI], arg);
4218 }
4219
4220 static void
4221 mdb_default_cmp(MDB_txn *txn, MDB_dbi dbi)
4222 {
4223         if (txn->mt_dbs[dbi].md_flags & MDB_REVERSEKEY)
4224                 txn->mt_dbxs[dbi].md_cmp = memnrcmp;
4225         else if (txn->mt_dbs[dbi].md_flags & MDB_INTEGERKEY)
4226                 txn->mt_dbxs[dbi].md_cmp = cintcmp;
4227         else
4228                 txn->mt_dbxs[dbi].md_cmp = memncmp;
4229
4230         if (txn->mt_dbs[dbi].md_flags & MDB_DUPSORT) {
4231                 if (txn->mt_dbs[dbi].md_flags & MDB_INTEGERDUP) {
4232                         if (txn->mt_dbs[dbi].md_flags & MDB_DUPFIXED)
4233                                 txn->mt_dbxs[dbi].md_dcmp = intcmp;
4234                         else
4235                                 txn->mt_dbxs[dbi].md_dcmp = cintcmp;
4236                 } else if (txn->mt_dbs[dbi].md_flags & MDB_REVERSEDUP) {
4237                         txn->mt_dbxs[dbi].md_dcmp = memnrcmp;
4238                 } else {
4239                         txn->mt_dbxs[dbi].md_dcmp = memncmp;
4240                 }
4241         } else {
4242                 txn->mt_dbxs[dbi].md_dcmp = NULL;
4243         }
4244 }
4245
4246 int mdb_open(MDB_txn *txn, const char *name, unsigned int flags, MDB_dbi *dbi)
4247 {
4248         MDB_val key, data;
4249         MDB_dbi i;
4250         int rc, dirty = 0;
4251         size_t len;
4252
4253         if (txn->mt_dbxs[FREE_DBI].md_cmp == NULL) {
4254                 mdb_default_cmp(txn, FREE_DBI);
4255         }
4256
4257         /* main DB? */
4258         if (!name) {
4259                 *dbi = MAIN_DBI;
4260                 if (flags & (MDB_DUPSORT|MDB_REVERSEKEY|MDB_INTEGERKEY))
4261                         txn->mt_dbs[MAIN_DBI].md_flags |= (flags & (MDB_DUPSORT|MDB_REVERSEKEY|MDB_INTEGERKEY));
4262                 mdb_default_cmp(txn, MAIN_DBI);
4263                 return MDB_SUCCESS;
4264         }
4265
4266         if (txn->mt_dbxs[MAIN_DBI].md_cmp == NULL) {
4267                 mdb_default_cmp(txn, MAIN_DBI);
4268         }
4269
4270         /* Is the DB already open? */
4271         len = strlen(name);
4272         for (i=2; i<txn->mt_numdbs; i++) {
4273                 if (len == txn->mt_dbxs[i].md_name.mv_size &&
4274                         !strncmp(name, txn->mt_dbxs[i].md_name.mv_data, len)) {
4275                         *dbi = i;
4276                         return MDB_SUCCESS;
4277                 }
4278         }
4279
4280         if (txn->mt_numdbs >= txn->mt_env->me_maxdbs - 1)
4281                 return ENFILE;
4282
4283         /* Find the DB info */
4284         key.mv_size = len;
4285         key.mv_data = (void *)name;
4286         rc = mdb_get(txn, MAIN_DBI, &key, &data);
4287
4288         /* Create if requested */
4289         if (rc == MDB_NOTFOUND && (flags & MDB_CREATE)) {
4290                 MDB_cursor mc;
4291                 MDB_db dummy;
4292                 data.mv_size = sizeof(MDB_db);
4293                 data.mv_data = &dummy;
4294                 memset(&dummy, 0, sizeof(dummy));
4295                 dummy.md_root = P_INVALID;
4296                 dummy.md_flags = flags & 0xffff;
4297                 mc.mc_txn = txn;
4298                 mc.mc_dbi = MAIN_DBI;
4299                 mc.mc_flags = 0;
4300                 rc = mdb_cursor_put(&mc, &key, &data, F_SUBDATA);
4301                 dirty = 1;
4302         }
4303
4304         /* OK, got info, add to table */
4305         if (rc == MDB_SUCCESS) {
4306                 txn->mt_dbxs[txn->mt_numdbs].md_name.mv_data = strdup(name);
4307                 txn->mt_dbxs[txn->mt_numdbs].md_name.mv_size = len;
4308                 txn->mt_dbxs[txn->mt_numdbs].md_rel = NULL;
4309                 txn->mt_dbxs[txn->mt_numdbs].md_parent = MAIN_DBI;
4310                 txn->mt_dbxs[txn->mt_numdbs].md_dirty = dirty;
4311                 memcpy(&txn->mt_dbs[txn->mt_numdbs], data.mv_data, sizeof(MDB_db));
4312                 *dbi = txn->mt_numdbs;
4313                 txn->mt_env->me_dbs[0][txn->mt_numdbs] = txn->mt_dbs[txn->mt_numdbs];
4314                 txn->mt_env->me_dbs[1][txn->mt_numdbs] = txn->mt_dbs[txn->mt_numdbs];
4315                 mdb_default_cmp(txn, txn->mt_numdbs);
4316                 txn->mt_numdbs++;
4317         }
4318
4319         return rc;
4320 }
4321
4322 int mdb_stat(MDB_txn *txn, MDB_dbi dbi, MDB_stat *arg)
4323 {
4324         if (txn == NULL || arg == NULL || dbi >= txn->mt_numdbs)
4325                 return EINVAL;
4326
4327         return mdb_stat0(txn->mt_env, &txn->mt_dbs[dbi], arg);
4328 }
4329
4330 void mdb_close(MDB_txn *txn, MDB_dbi dbi)
4331 {
4332         char *ptr;
4333         if (dbi <= MAIN_DBI || dbi >= txn->mt_numdbs)
4334                 return;
4335         ptr = txn->mt_dbxs[dbi].md_name.mv_data;
4336         txn->mt_dbxs[dbi].md_name.mv_data = NULL;
4337         txn->mt_dbxs[dbi].md_name.mv_size = 0;
4338         free(ptr);
4339 }
4340
4341 int mdb_set_compare(MDB_txn *txn, MDB_dbi dbi, MDB_cmp_func *cmp)
4342 {
4343         if (txn == NULL || !dbi || dbi >= txn->mt_numdbs)
4344                 return EINVAL;
4345
4346         txn->mt_dbxs[dbi].md_cmp = cmp;
4347         return MDB_SUCCESS;
4348 }
4349
4350 int mdb_set_dupsort(MDB_txn *txn, MDB_dbi dbi, MDB_cmp_func *cmp)
4351 {
4352         if (txn == NULL || !dbi || dbi >= txn->mt_numdbs)
4353                 return EINVAL;
4354
4355         txn->mt_dbxs[dbi].md_dcmp = cmp;
4356         return MDB_SUCCESS;
4357 }
4358
4359 int mdb_set_relfunc(MDB_txn *txn, MDB_dbi dbi, MDB_rel_func *rel)
4360 {
4361         if (txn == NULL || !dbi || dbi >= txn->mt_numdbs)
4362                 return EINVAL;
4363
4364         txn->mt_dbxs[dbi].md_rel = rel;
4365         return MDB_SUCCESS;
4366 }
4367
4368 /** @} */