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

   1 #include <sys/types.h>
   2 #include <sys/stat.h>
   3 #include <sys/queue.h>
   4 #include <sys/param.h>
   5 #include <sys/uio.h>
   6 #include <sys/mman.h>
   7 #ifdef HAVE_SYS_FILE_H
   8 #include <sys/file.h>
   9 #endif
  10 #include <fcntl.h>
  11
  12 #include <assert.h>
  13 #include <err.h>
  14 #include <errno.h>
  15 #include <fcntl.h>
  16 #include <stddef.h>
  17 #include <stdint.h>
  18 #include <stdio.h>
  19 #include <stdlib.h>
  20 #include <string.h>
  21 #include <time.h>
  22 #include <unistd.h>
  23 #include <pthread.h>
  24
  25 #include "mdb.h"
  26
  27 typedef ulong            pgno_t;
  28
  29 #include "idl.h"
  30
  31 #ifndef DEBUG
  32 #define DEBUG 1
  33 #endif
  34
  35 #if (DEBUG +0) && defined(__GNUC__)
  36 # define DPRINTF(fmt, ...) \
  37         fprintf(stderr, "%s:%d: " fmt "\n", __func__, __LINE__, ##__VA_ARGS__)
  38 #else
  39 # define DPRINTF(...)   ((void) 0)
  40 #endif
  41
  42 #define PAGESIZE         4096
  43 #define MDB_MINKEYS      4
  44 #define MDB_MAGIC        0xBEEFC0DE
  45 #define MDB_VERSION      1
  46 #define MAXKEYSIZE       255
  47
  48 #define P_INVALID        (~0UL)
  49
  50 #define F_ISSET(w, f)    (((w) & (f)) == (f))
  51
  52 typedef uint16_t         indx_t;
  53
  54 #define DEFAULT_READERS 126
  55 #define DEFAULT_MAPSIZE 1048576
  56
  57 /* Lock descriptor stuff */
  58 #define RXBODY  \
  59         ulong           mr_txnid; \
  60         pid_t           mr_pid; \
  61         pthread_t       mr_tid
  62 typedef struct MDB_rxbody {
  63         RXBODY;
  64 } MDB_rxbody;
  65
  66 #ifndef CACHELINE
  67 #define CACHELINE       64      /* most CPUs. Itanium uses 128 */
  68 #endif
  69
  70 typedef struct MDB_reader {
  71         RXBODY;
  72         /* cache line alignment */
  73         char pad[CACHELINE-sizeof(MDB_rxbody)];
  74 } MDB_reader;
  75
  76 #define TXBODY \
  77         uint32_t        mt_magic;       \
  78         uint32_t        mt_version;     \
  79         pthread_mutex_t mt_mutex;       \
  80         ulong           mt_txnid;       \
  81         uint32_t        mt_numreaders
  82 typedef struct MDB_txbody {
  83         TXBODY;
  84 } MDB_txbody;
  85
  86 typedef struct MDB_txninfo {
  87         TXBODY;
  88         char pad[CACHELINE-sizeof(MDB_txbody)];
  89         pthread_mutex_t mt_wmutex;
  90         char pad2[CACHELINE-sizeof(pthread_mutex_t)];
  91         MDB_reader      mt_readers[1];
  92 } MDB_txninfo;
  93
  94 /* Common header for all page types. Overflow pages
  95  * occupy a number of contiguous pages with no
  96  * headers on any page after the first.
  97  */
  98 typedef struct MDB_page {               /* represents a page of storage */
  99         pgno_t          mp_pgno;                /* page number */
 100 #define P_BRANCH         0x01           /* branch page */
 101 #define P_LEAF           0x02           /* leaf page */
 102 #define P_OVERFLOW       0x04           /* overflow page */
 103 #define P_META           0x08           /* meta page */
 104 #define P_DIRTY          0x10           /* dirty page */
 105         uint32_t        mp_flags;
 106 #define mp_lower        mp_pb.pb.pb_lower
 107 #define mp_upper        mp_pb.pb.pb_upper
 108 #define mp_pages        mp_pb.pb_pages
 109         union page_bounds {
 110                 struct {
 111                         indx_t          pb_lower;               /* lower bound of free space */
 112                         indx_t          pb_upper;               /* upper bound of free space */
 113                 } pb;
 114                 uint32_t        pb_pages;       /* number of overflow pages */
 115         } mp_pb;
 116         indx_t          mp_ptrs[1];             /* dynamic size */
 117 } MDB_page;
 118
 119 #define PAGEHDRSZ        ((unsigned) offsetof(MDB_page, mp_ptrs))
 120
 121 #define NUMKEYS(p)       (((p)->mp_lower - PAGEHDRSZ) >> 1)
 122 #define SIZELEFT(p)      (indx_t)((p)->mp_upper - (p)->mp_lower)
 123 #define PAGEFILL(env, p) (1000L * ((env)->me_meta.mm_stat.ms_psize - PAGEHDRSZ - SIZELEFT(p)) / \
 124                                 ((env)->me_meta.mm_stat.ms_psize - PAGEHDRSZ))
 125 #define IS_LEAF(p)       F_ISSET((p)->mp_flags, P_LEAF)
 126 #define IS_BRANCH(p)     F_ISSET((p)->mp_flags, P_BRANCH)
 127 #define IS_OVERFLOW(p)   F_ISSET((p)->mp_flags, P_OVERFLOW)
 128
 129 #define OVPAGES(size, psize)    (PAGEHDRSZ + size + psize - 1) / psize;
 130
 131 typedef struct MDB_meta {                       /* meta (footer) page content */
 132         uint32_t        mm_magic;
 133         uint32_t        mm_version;
 134         void            *mm_address;            /* address for fixed mapping */
 135         size_t          mm_mapsize;                     /* size of mmap region */
 136         pgno_t          mm_last_pg;                     /* last used page in file */
 137         ulong           mm_txnid;                       /* txnid that committed this page */
 138         MDB_stat        mm_stat;
 139         pgno_t          mm_root;                        /* page number of root page */
 140 } MDB_meta;
 141
 142 typedef struct MDB_dhead {                                      /* a dirty page */
 143         SIMPLEQ_ENTRY(MDB_dpage)         md_next;       /* queue of dirty pages */
 144         MDB_page        *md_parent;
 145         unsigned        md_pi;                          /* parent index */
 146         int                     md_num;
 147 } MDB_dhead;
 148
 149 typedef struct MDB_dpage {
 150         MDB_dhead       h;
 151         MDB_page        p;
 152 } MDB_dpage;
 153
 154 SIMPLEQ_HEAD(dirty_queue, MDB_dpage);
 155
 156 typedef struct MDB_oldpages {
 157         struct MDB_oldpages *mo_next;
 158         ulong           mo_txnid;
 159         pgno_t          mo_pages[1];    /* dynamic */
 160 } MDB_oldpages;
 161
 162 typedef struct MDB_pageparent {
 163         MDB_page *mp_page;
 164         MDB_page *mp_parent;
 165         unsigned mp_pi;
 166 } MDB_pageparent;
 167
 168 static MDB_dpage *mdb_newpage(MDB_txn *txn, MDB_page *parent, unsigned int parent_idx, int num);
 169 static int              mdb_touch(MDB_txn *txn, MDB_pageparent *mp);
 170
 171 typedef struct MDB_ppage {                                      /* ordered list of pages */
 172         SLIST_ENTRY(MDB_ppage)   mp_entry;
 173         MDB_page                *mp_page;
 174         unsigned int    mp_ki;          /* cursor index on page */
 175 } MDB_ppage;
 176 SLIST_HEAD(page_stack, MDB_ppage);
 177
 178 #define CURSOR_EMPTY(c)          SLIST_EMPTY(&(c)->mc_stack)
 179 #define CURSOR_TOP(c)            SLIST_FIRST(&(c)->mc_stack)
 180 #define CURSOR_POP(c)            SLIST_REMOVE_HEAD(&(c)->mc_stack, mp_entry)
 181 #define CURSOR_PUSH(c,p)         SLIST_INSERT_HEAD(&(c)->mc_stack, p, mp_entry)
 182
 183 struct MDB_cursor {
 184         MDB_db          *mc_db;
 185         MDB_txn         *mc_txn;
 186         struct page_stack        mc_stack;              /* stack of parent pages */
 187         short           mc_initialized; /* 1 if initialized */
 188         short           mc_eof;         /* 1 if end is reached */
 189 };
 190
 191 #define METAHASHLEN      offsetof(MDB_meta, mm_hash)
 192 #define METADATA(p)      ((void *)((char *)p + PAGEHDRSZ))
 193
 194 typedef struct MDB_node {
 195 #define mn_pgno          mn_p.np_pgno
 196 #define mn_dsize         mn_p.np_dsize
 197         union {
 198                 pgno_t           np_pgno;       /* child page number */
 199                 uint32_t         np_dsize;      /* leaf data size */
 200         } mn_p;
 201         unsigned int    mn_flags:4;
 202         unsigned int    mn_ksize:12;                    /* key size */
 203 #define F_BIGDATA        0x01                   /* data put on overflow page */
 204         char            mn_data[1];
 205 } MDB_node;
 206
 207 struct MDB_txn {
 208         pgno_t          mt_root;                /* current / new root page */
 209         pgno_t          mt_next_pgno;   /* next unallocated page */
 210         ulong           mt_txnid;
 211         ulong           mt_oldest;
 212         MDB_env         *mt_env;
 213         pgno_t          *mt_free_pgs;   /* this is an IDL */
 214         union {
 215                 struct dirty_queue      *dirty_queue;   /* modified pages */
 216                 MDB_reader      *reader;
 217         } mt_u;
 218 #define MDB_TXN_RDONLY           0x01           /* read-only transaction */
 219 #define MDB_TXN_ERROR            0x02           /* an error has occurred */
 220 #define MDB_TXN_METOGGLE        0x04            /* used meta page 1 */
 221         unsigned int             mt_flags;
 222 };
 223
 224 struct MDB_db {
 225         unsigned int    md_flags;
 226         MDB_cmp_func    *md_cmp;                /* user compare function */
 227         MDB_rel_func    *md_rel;                /* user relocate function */
 228         MDB_db                  *md_parent;             /* parent tree */
 229         MDB_env                 *md_env;
 230         MDB_stat                md_stat;
 231         pgno_t                  md_root;                /* page number of root page */
 232 };
 233
 234 struct MDB_env {
 235         int                     me_fd;
 236         int                     me_lfd;
 237         uint32_t        me_flags;
 238         unsigned int                    me_maxreaders;
 239         char            *me_path;
 240         char *me_map;
 241         MDB_txninfo     *me_txns;
 242         MDB_db          me_db;          /* first DB */
 243         MDB_meta        me_meta;
 244         MDB_txn *me_txn;                /* current write transaction */
 245         size_t          me_mapsize;
 246         off_t           me_size;                /* current file size */
 247         pthread_key_t   me_txkey;       /* thread-key for readers */
 248         MDB_oldpages *me_pghead;
 249         MDB_oldpages *me_pgtail;
 250 };
 251
 252 #define NODESIZE         offsetof(MDB_node, mn_data)
 253
 254 #define INDXSIZE(k)      (NODESIZE + ((k) == NULL ? 0 : (k)->mv_size))
 255 #define LEAFSIZE(k, d)   (NODESIZE + (k)->mv_size + (d)->mv_size)
 256 #define NODEPTR(p, i)    ((MDB_node *)((char *)(p) + (p)->mp_ptrs[i]))
 257 #define NODEKEY(node)    (void *)((node)->mn_data)
 258 #define NODEDATA(node)   (void *)((char *)(node)->mn_data + (node)->mn_ksize)
 259 #define NODEPGNO(node)   ((node)->mn_pgno)
 260 #define NODEDSZ(node)    ((node)->mn_dsize)
 261
 262 #define MDB_COMMIT_PAGES         64     /* max number of pages to write in one commit */
 263 #define MDB_MAXCACHE_DEF         1024   /* max number of pages to keep in cache  */
 264
 265 static int  mdb_search_page_root(MDB_db *db,
 266                             MDB_val *key,
 267                             MDB_cursor *cursor, int modify,
 268                             MDB_pageparent *mpp);
 269 static int  mdb_search_page(MDB_db *db,
 270                             MDB_txn *txn, MDB_val *key,
 271                             MDB_cursor *cursor, int modify,
 272                             MDB_pageparent *mpp);
 273
 274 static int  mdbenv_read_header(MDB_env *env);
 275 static int  mdb_check_meta_page(MDB_page *p);
 276 static int  mdbenv_read_meta(MDB_env *env, int *which);
 277 static int  mdbenv_write_meta(MDB_txn *txn);
 278 static MDB_page *mdbenv_get_page(MDB_env *env, pgno_t pgno);
 279
 280 static MDB_node *mdb_search_node(MDB_db *db, MDB_page *mp,
 281                             MDB_val *key, int *exactp, unsigned int *kip);
 282 static int  mdb_add_node(MDB_db *bt, MDB_page *mp,
 283                             indx_t indx, MDB_val *key, MDB_val *data,
 284                             pgno_t pgno, uint8_t flags);
 285 static void mdb_del_node(MDB_db *bt, MDB_page *mp,
 286                             indx_t indx);
 287 static int  mdb_read_data(MDB_db *bt, MDB_page *mp,
 288                             MDB_node *leaf, MDB_val *data);
 289
 290 static int               mdb_rebalance(MDB_db *bt, MDB_pageparent *mp);
 291 static int               mdb_update_key(MDB_db *bt, MDB_page *mp,
 292                             indx_t indx, MDB_val *key);
 293 static int               mdb_move_node(MDB_db *bt,
 294                                 MDB_pageparent *src, indx_t srcindx,
 295                                 MDB_pageparent *dst, indx_t dstindx);
 296 static int               mdb_merge(MDB_db *bt, MDB_pageparent *src,
 297                             MDB_pageparent *dst);
 298 static int               mdb_split(MDB_db *bt, MDB_page **mpp,
 299                             unsigned int *newindxp, MDB_val *newkey,
 300                             MDB_val *newdata, pgno_t newpgno);
 301 static MDB_dpage *mdbenv_new_page(MDB_env *env, uint32_t flags, int num);
 302
 303 static void              cursor_pop_page(MDB_cursor *cursor);
 304 static MDB_ppage *cursor_push_page(MDB_cursor *cursor,
 305                             MDB_page *mp);
 306
 307 static int               mdb_set_key(MDB_db *bt, MDB_page *mp,
 308                             MDB_node *node, MDB_val *key);
 309 static int               mdb_sibling(MDB_cursor *cursor, int move_right);
 310 static int               mdb_cursor_next(MDB_cursor *cursor,
 311                             MDB_val *key, MDB_val *data);
 312 static int               mdb_cursor_set(MDB_cursor *cursor,
 313                             MDB_val *key, MDB_val *data, int *exactp);
 314 static int               mdb_cursor_first(MDB_cursor *cursor,
 315                             MDB_val *key, MDB_val *data);
 316
 317 static size_t            mdb_leaf_size(MDB_db *bt, MDB_val *key,
 318                             MDB_val *data);
 319 static size_t            mdb_branch_size(MDB_db *bt, MDB_val *key);
 320
 321 static int               memncmp(const void *s1, size_t n1,
 322                                  const void *s2, size_t n2);
 323 static int               memnrcmp(const void *s1, size_t n1,
 324                                   const void *s2, size_t n2);
 325
 326 static int
 327 memncmp(const void *s1, size_t n1, const void *s2, size_t n2)
 328 {
 329         int diff, len_diff = -1;
 330
 331         if (n1 >= n2) {
 332                 len_diff = (n1 > n2);
 333                 n1 = n2;
 334         }
 335         diff = memcmp(s1, s2, n1);
 336         return diff ? diff : len_diff;
 337 }
 338
 339 static int
 340 memnrcmp(const void *s1, size_t n1, const void *s2, size_t n2)
 341 {
 342         const unsigned char     *p1, *p2, *p1_lim;
 343
 344         if (n2 == 0)
 345                 return n1 != 0;
 346         if (n1 == 0)
 347                 return -1;
 348
 349         p1 = (const unsigned char *)s1 + n1 - 1;
 350         p2 = (const unsigned char *)s2 + n2 - 1;
 351
 352         for (p1_lim = (n1 <= n2 ? s1 : s2);  *p1 == *p2;  p1--, p2--) {
 353                 if (p1 == p1_lim)
 354                         return (p1 != s1) ? (p1 != p2) : (p2 != s2) ? -1 : 0;
 355         }
 356         return *p1 - *p2;
 357 }
 358
 359 int
 360 mdb_cmp(MDB_db *db, const MDB_val *a, const MDB_val *b)
 361 {
 362         return db->md_cmp(a, b);
 363 }
 364
 365 static int
 366 _mdb_cmp(MDB_db *db, const MDB_val *key1, const MDB_val *key2)
 367 {
 368         if (F_ISSET(db->md_flags, MDB_REVERSEKEY))
 369                 return memnrcmp(key1->mv_data, key1->mv_size, key2->mv_data, key2->mv_size);
 370         else
 371                 return memncmp((char *)key1->mv_data, key1->mv_size, key2->mv_data, key2->mv_size);
 372 }
 373
 374 /* Allocate new page(s) for writing */
 375 static MDB_dpage *
 376 mdb_newpage(MDB_txn *txn, MDB_page *parent, unsigned int parent_idx, int num)
 377 {
 378         MDB_dpage *dp;
 379         pgno_t pgno = P_INVALID;
 380
 381         if (txn->mt_env->me_pghead) {
 382                 ulong oldest = txn->mt_txnid - 2;
 383                 unsigned int i;
 384                 for (i=0; i<txn->mt_env->me_txns->mt_numreaders; i++) {
 385                         if (txn->mt_env->me_txns->mt_readers[i].mr_txnid < oldest)
 386                                 oldest = txn->mt_env->me_txns->mt_readers[i].mr_txnid;
 387                 }
 388                 if (oldest > txn->mt_env->me_pghead->mo_txnid) {
 389                         MDB_oldpages *mop = txn->mt_env->me_pghead;
 390                         txn->mt_oldest = oldest;
 391                         if (num > 1) {
 392                                 /* FIXME */
 393                                 ;
 394                         } else {
 395                                 /* peel pages off tail, so we only have to truncate the list */
 396                                 pgno = MDB_IDL_LAST(mop->mo_pages);
 397                                 if (MDB_IDL_IS_RANGE(mop->mo_pages)) {
 398                                         mop->mo_pages[2]++;
 399                                         if (mop->mo_pages[2] > mop->mo_pages[1])
 400                                                 mop->mo_pages[0] = 0;
 401                                 } else {
 402                                         mop->mo_pages[0]--;
 403                                 }
 404                                 if (MDB_IDL_IS_ZERO(mop->mo_pages)) {
 405                                         txn->mt_env->me_pghead = mop->mo_next;
 406                                         if (!txn->mt_env->me_pghead)
 407                                                 txn->mt_env->me_pgtail = NULL;
 408                                         free(mop);
 409                                 }
 410                         }
 411                 }
 412         }
 413
 414         if ((dp = malloc(txn->mt_env->me_meta.mm_stat.ms_psize * num + sizeof(MDB_dhead))) == NULL)
 415                 return NULL;
 416         dp->h.md_num = num;
 417         dp->h.md_parent = parent;
 418         dp->h.md_pi = parent_idx;
 419         SIMPLEQ_INSERT_TAIL(txn->mt_u.dirty_queue, dp, h.md_next);
 420         if (pgno == P_INVALID) {
 421                 dp->p.mp_pgno = txn->mt_next_pgno;
 422                 txn->mt_next_pgno += num;
 423         } else {
 424                 dp->p.mp_pgno = pgno;
 425         }
 426
 427         return dp;
 428 }
 429
 430 /* Touch a page: make it dirty and re-insert into tree with updated pgno.
 431  */
 432 static int
 433 mdb_touch(MDB_txn *txn, MDB_pageparent *pp)
 434 {
 435         MDB_page *mp = pp->mp_page;
 436         pgno_t  pgno;
 437         assert(txn != NULL);
 438         assert(pp != NULL);
 439
 440         if (!F_ISSET(mp->mp_flags, P_DIRTY)) {
 441                 MDB_dpage *dp;
 442                 if ((dp = mdb_newpage(txn, pp->mp_parent, pp->mp_pi, 1)) == NULL)
 443                         return ENOMEM;
 444                 DPRINTF("touched page %lu -> %lu", mp->mp_pgno, dp->p.mp_pgno);
 445                 mdb_idl_insert(txn->mt_free_pgs, mp->mp_pgno);
 446                 pgno = dp->p.mp_pgno;
 447                 bcopy(mp, &dp->p, txn->mt_env->me_meta.mm_stat.ms_psize);
 448                 mp = &dp->p;
 449                 mp->mp_pgno = pgno;
 450                 mp->mp_flags |= P_DIRTY;
 451
 452                 /* Update the page number to new touched page. */
 453                 if (pp->mp_parent != NULL)
 454                         NODEPGNO(NODEPTR(pp->mp_parent, pp->mp_pi)) = mp->mp_pgno;
 455                 pp->mp_page = mp;
 456         }
 457         return 0;
 458 }
 459
 460 int
 461 mdbenv_sync(MDB_env *env)
 462 {
 463         int rc = 0;
 464         if (!F_ISSET(env->me_flags, MDB_NOSYNC)) {
 465                 if (fsync(env->me_fd))
 466                         rc = errno;
 467         }
 468         return rc;
 469 }
 470
 471 int
 472 mdb_txn_begin(MDB_env *env, int rdonly, MDB_txn **ret)
 473 {
 474         MDB_txn *txn;
 475         int rc, toggle;
 476
 477         if ((txn = calloc(1, sizeof(*txn))) == NULL) {
 478                 DPRINTF("calloc: %s", strerror(errno));
 479                 return ENOMEM;
 480         }
 481
 482         if (rdonly) {
 483                 txn->mt_flags |= MDB_TXN_RDONLY;
 484         } else {
 485                 txn->mt_u.dirty_queue = calloc(1, sizeof(*txn->mt_u.dirty_queue));
 486                 if (txn->mt_u.dirty_queue == NULL) {
 487                         free(txn);
 488                         return ENOMEM;
 489                 }
 490                 SIMPLEQ_INIT(txn->mt_u.dirty_queue);
 491
 492                 pthread_mutex_lock(&env->me_txns->mt_wmutex);
 493                 env->me_txns->mt_txnid++;
 494                 txn->mt_free_pgs = malloc(MDB_IDL_UM_SIZEOF);
 495                 if (txn->mt_free_pgs == NULL) {
 496                         free(txn->mt_u.dirty_queue);
 497                         free(txn);
 498                         return ENOMEM;
 499                 }
 500                 txn->mt_free_pgs[0] = 0;
 501         }
 502         txn->mt_txnid = env->me_txns->mt_txnid;
 503         if (rdonly) {
 504                 MDB_reader *r = pthread_getspecific(env->me_txkey);
 505                 if (!r) {
 506                         unsigned int i;
 507                         pthread_mutex_lock(&env->me_txns->mt_mutex);
 508                         for (i=0; i<env->me_maxreaders; i++) {
 509                                 if (env->me_txns->mt_readers[i].mr_pid == 0) {
 510                                         env->me_txns->mt_readers[i].mr_pid = getpid();
 511                                         env->me_txns->mt_readers[i].mr_tid = pthread_self();
 512                                         pthread_setspecific(env->me_txkey, &env->me_txns->mt_readers[i]);
 513                                         if (i >= env->me_txns->mt_numreaders)
 514                                                 env->me_txns->mt_numreaders = i+1;
 515                                         break;
 516                                 }
 517                         }
 518                         pthread_mutex_unlock(&env->me_txns->mt_mutex);
 519                         if (i == env->me_maxreaders) {
 520                                 return ENOSPC;
 521                         }
 522                 }
 523                 r->mr_txnid = txn->mt_txnid;
 524                 txn->mt_u.reader = r;
 525         } else {
 526                 env->me_txn = txn;
 527         }
 528
 529         txn->mt_env = env;
 530
 531         if ((rc = mdbenv_read_meta(env, &toggle)) != MDB_SUCCESS) {
 532                 mdb_txn_abort(txn);
 533                 return rc;
 534         }
 535
 536         if (toggle)
 537                 txn->mt_flags |= MDB_TXN_METOGGLE;
 538
 539         txn->mt_next_pgno = env->me_meta.mm_last_pg+1;
 540         txn->mt_root = env->me_meta.mm_root;
 541         DPRINTF("begin transaction %lu on mdbenv %p, root page %lu",
 542                 txn->mt_txnid, (void *) env, txn->mt_root);
 543
 544         *ret = txn;
 545         return MDB_SUCCESS;
 546 }
 547
 548 void
 549 mdb_txn_abort(MDB_txn *txn)
 550 {
 551         MDB_dpage *dp;
 552         MDB_env *env;
 553
 554         if (txn == NULL)
 555                 return;
 556
 557         env = txn->mt_env;
 558         DPRINTF("abort transaction %lu on mdbenv %p, root page %lu",
 559                 txn->mt_txnid, (void *) env, txn->mt_root);
 560
 561         if (F_ISSET(txn->mt_flags, MDB_TXN_RDONLY)) {
 562                 txn->mt_u.reader->mr_txnid = 0;
 563         } else {
 564                 /* Discard all dirty pages. Return any re-used pages
 565                  * to the free list.
 566                  */
 567                 MDB_IDL_ZERO(txn->mt_free_pgs);
 568                 while (!SIMPLEQ_EMPTY(txn->mt_u.dirty_queue)) {
 569                         dp = SIMPLEQ_FIRST(txn->mt_u.dirty_queue);
 570                         SIMPLEQ_REMOVE_HEAD(txn->mt_u.dirty_queue, h.md_next);
 571                         if (dp->p.mp_pgno <= env->me_meta.mm_last_pg)
 572                                 mdb_idl_insert(txn->mt_free_pgs, dp->p.mp_pgno);
 573                         free(dp);
 574                 }
 575                 /* put back to head of free list */
 576                 if (!MDB_IDL_IS_ZERO(txn->mt_free_pgs)) {
 577                         MDB_oldpages *mop;
 578
 579                         mop = malloc(sizeof(MDB_oldpages) + MDB_IDL_SIZEOF(txn->mt_free_pgs) - sizeof(pgno_t));
 580                         mop->mo_next = env->me_pghead;
 581                         mop->mo_txnid = txn->mt_oldest - 1;
 582                         if (!env->me_pghead) {
 583                                 env->me_pgtail = mop;
 584                         }
 585                         env->me_pghead = mop;
 586                         bcopy(txn->mt_free_pgs, mop->mo_pages, MDB_IDL_SIZEOF(txn->mt_free_pgs));
 587                 }
 588
 589                 free(txn->mt_free_pgs);
 590                 free(txn->mt_u.dirty_queue);
 591                 env->me_txn = NULL;
 592                 env->me_txns->mt_txnid--;
 593                 pthread_mutex_unlock(&env->me_txns->mt_wmutex);
 594         }
 595
 596         free(txn);
 597 }
 598
 599 int
 600 mdb_txn_commit(MDB_txn *txn)
 601 {
 602         int              n, done;
 603         ssize_t          rc;
 604         off_t            size;
 605         MDB_dpage       *dp;
 606         MDB_env *env;
 607         pgno_t  next;
 608         struct iovec     iov[MDB_COMMIT_PAGES];
 609
 610         assert(txn != NULL);
 611         assert(txn->mt_env != NULL);
 612
 613         env = txn->mt_env;
 614
 615         if (F_ISSET(txn->mt_flags, MDB_TXN_RDONLY)) {
 616                 DPRINTF("attempt to commit read-only transaction");
 617                 mdb_txn_abort(txn);
 618                 return EPERM;
 619         }
 620
 621         if (txn != env->me_txn) {
 622                 DPRINTF("attempt to commit unknown transaction");
 623                 mdb_txn_abort(txn);
 624                 return EINVAL;
 625         }
 626
 627         if (F_ISSET(txn->mt_flags, MDB_TXN_ERROR)) {
 628                 DPRINTF("error flag is set, can't commit");
 629                 mdb_txn_abort(txn);
 630                 return EINVAL;
 631         }
 632
 633         if (SIMPLEQ_EMPTY(txn->mt_u.dirty_queue))
 634                 goto done;
 635
 636         DPRINTF("committing transaction %lu on mdbenv %p, root page %lu",
 637             txn->mt_txnid, (void *) env, txn->mt_root);
 638
 639         /* Commit up to MDB_COMMIT_PAGES dirty pages to disk until done.
 640          */
 641         next = 0;
 642         do {
 643                 n = 0;
 644                 done = 1;
 645                 size = 0;
 646                 SIMPLEQ_FOREACH(dp, txn->mt_u.dirty_queue, h.md_next) {
 647                         if (dp->p.mp_pgno != next) {
 648                                 if (n) {
 649                                         DPRINTF("committing %u dirty pages", n);
 650                                         rc = writev(env->me_fd, iov, n);
 651                                         if (rc != size) {
 652                                                 n = errno;
 653                                                 if (rc > 0)
 654                                                         DPRINTF("short write, filesystem full?");
 655                                                 else
 656                                                         DPRINTF("writev: %s", strerror(errno));
 657                                                 mdb_txn_abort(txn);
 658                                                 return n;
 659                                         }
 660                                         n = 0;
 661                                         size = 0;
 662                                 }
 663                                 lseek(env->me_fd, dp->p.mp_pgno * env->me_meta.mm_stat.ms_psize, SEEK_SET);
 664                                 next = dp->p.mp_pgno;
 665                         }
 666                         DPRINTF("committing page %lu", dp->p.mp_pgno);
 667                         iov[n].iov_len = env->me_meta.mm_stat.ms_psize * dp->h.md_num;
 668                         iov[n].iov_base = &dp->p;
 669                         size += iov[n].iov_len;
 670                         next = dp->p.mp_pgno + dp->h.md_num;
 671                         /* clear dirty flag */
 672                         dp->p.mp_flags &= ~P_DIRTY;
 673                         if (++n >= MDB_COMMIT_PAGES) {
 674                                 done = 0;
 675                                 break;
 676                         }
 677                 }
 678
 679                 if (n == 0)
 680                         break;
 681
 682                 DPRINTF("committing %u dirty pages", n);
 683                 rc = writev(env->me_fd, iov, n);
 684                 if (rc != size) {
 685                         n = errno;
 686                         if (rc > 0)
 687                                 DPRINTF("short write, filesystem full?");
 688                         else
 689                                 DPRINTF("writev: %s", strerror(errno));
 690                         mdb_txn_abort(txn);
 691                         return n;
 692                 }
 693
 694         } while (!done);
 695
 696         /* Drop the dirty pages.
 697          */
 698         while (!SIMPLEQ_EMPTY(txn->mt_u.dirty_queue)) {
 699                 dp = SIMPLEQ_FIRST(txn->mt_u.dirty_queue);
 700                 SIMPLEQ_REMOVE_HEAD(txn->mt_u.dirty_queue, h.md_next);
 701                 free(dp);
 702         }
 703
 704         if ((n = mdbenv_sync(env)) != 0 ||
 705             (n = mdbenv_write_meta(txn)) != MDB_SUCCESS ||
 706             (n = mdbenv_sync(env)) != 0) {
 707                 mdb_txn_abort(txn);
 708                 return n;
 709         }
 710         env->me_txn = NULL;
 711
 712         /* add to tail of free list */
 713         if (!MDB_IDL_IS_ZERO(txn->mt_free_pgs)) {
 714                 MDB_oldpages *mop;
 715
 716                 mop = malloc(sizeof(MDB_oldpages) + MDB_IDL_SIZEOF(txn->mt_free_pgs) - sizeof(pgno_t));
 717                 mop->mo_next = NULL;
 718                 if (env->me_pghead) {
 719                         env->me_pgtail->mo_next = mop;
 720                 } else {
 721                         env->me_pghead = mop;
 722                 }
 723                 env->me_pgtail = mop;
 724                 bcopy(txn->mt_free_pgs, mop->mo_pages, MDB_IDL_SIZEOF(txn->mt_free_pgs));
 725                 mop->mo_txnid = txn->mt_txnid;
 726         }
 727
 728         pthread_mutex_unlock(&env->me_txns->mt_wmutex);
 729         free(txn->mt_free_pgs);
 730         free(txn->mt_u.dirty_queue);
 731         free(txn);
 732         txn = NULL;
 733
 734 done:
 735         mdb_txn_abort(txn);
 736
 737         return MDB_SUCCESS;
 738 }
 739
 740 static int
 741 mdbenv_read_header(MDB_env *env)
 742 {
 743         char             page[PAGESIZE];
 744         MDB_page        *p;
 745         MDB_meta        *m;
 746         int              rc;
 747
 748         assert(env != NULL);
 749
 750         /* We don't know the page size yet, so use a minimum value.
 751          */
 752
 753         if ((rc = pread(env->me_fd, page, PAGESIZE, 0)) == 0) {
 754                 return ENOENT;
 755         } else if (rc != PAGESIZE) {
 756                 if (rc > 0)
 757                         errno = EINVAL;
 758                 DPRINTF("read: %s", strerror(errno));
 759                 return errno;
 760         }
 761
 762         p = (MDB_page *)page;
 763
 764         if (!F_ISSET(p->mp_flags, P_META)) {
 765                 DPRINTF("page %lu not a meta page", p->mp_pgno);
 766                 return EINVAL;
 767         }
 768
 769         m = METADATA(p);
 770         if (m->mm_magic != MDB_MAGIC) {
 771                 DPRINTF("meta has invalid magic");
 772                 return EINVAL;
 773         }
 774
 775         if (m->mm_version != MDB_VERSION) {
 776                 DPRINTF("database is version %u, expected version %u",
 777                     m->mm_version, MDB_VERSION);
 778                 return EINVAL;
 779         }
 780
 781         bcopy(m, &env->me_meta, sizeof(*m));
 782         return 0;
 783 }
 784
 785 static int
 786 mdbenv_init_meta(MDB_env *env)
 787 {
 788         MDB_page *p, *q;
 789         MDB_meta *meta;
 790         int rc;
 791         unsigned int     psize;
 792
 793         DPRINTF("writing new meta page");
 794         psize = sysconf(_SC_PAGE_SIZE);
 795
 796         env->me_meta.mm_magic = MDB_MAGIC;
 797         env->me_meta.mm_version = MDB_VERSION;
 798         env->me_meta.mm_stat.ms_psize = psize;
 799         env->me_meta.mm_stat.ms_flags = env->me_flags & 0xffff;
 800         env->me_meta.mm_root = P_INVALID;
 801         env->me_meta.mm_last_pg = 1;
 802
 803         p = calloc(2, psize);
 804         p->mp_pgno = 0;
 805         p->mp_flags = P_META;
 806
 807         meta = METADATA(p);
 808         bcopy(&env->me_meta, meta, sizeof(*meta));
 809
 810         q = (MDB_page *)((char *)p + psize);
 811
 812         q->mp_pgno = 1;
 813         q->mp_flags = P_META;
 814
 815         meta = METADATA(q);
 816         bcopy(&env->me_meta, meta, sizeof(*meta));
 817
 818         rc = write(env->me_fd, p, psize * 2);
 819         free(p);
 820         return (rc == (int)psize * 2) ? MDB_SUCCESS : errno;
 821 }
 822
 823 static int
 824 mdbenv_write_meta(MDB_txn *txn)
 825 {
 826         MDB_env *env;
 827         MDB_meta        meta;
 828         off_t off;
 829         int rc;
 830
 831         assert(txn != NULL);
 832         assert(txn->mt_env != NULL);
 833
 834         DPRINTF("writing meta page for root page %lu", txn->mt_root);
 835
 836         env = txn->mt_env;
 837
 838         bcopy(&env->me_meta, &meta, sizeof(meta));
 839         meta.mm_root = txn->mt_root;
 840         meta.mm_last_pg = txn->mt_next_pgno - 1;
 841         meta.mm_txnid = txn->mt_txnid;
 842
 843         off = 0;
 844         if (!F_ISSET(txn->mt_flags, MDB_TXN_METOGGLE))
 845                 off = env->me_meta.mm_stat.ms_psize;
 846         off += PAGEHDRSZ;
 847
 848         lseek(env->me_fd, off, SEEK_SET);
 849         rc = write(env->me_fd, &meta, sizeof(meta));
 850         if (rc != sizeof(meta)) {
 851                 DPRINTF("write failed, disk error?");
 852                 return errno;
 853         }
 854
 855         return MDB_SUCCESS;
 856 }
 857
 858 /* Returns true if page p is a valid meta page, false otherwise.
 859  */
 860 static int
 861 mdb_check_meta_page(MDB_page *p)
 862 {
 863         if (!F_ISSET(p->mp_flags, P_META)) {
 864                 DPRINTF("page %lu not a meta page", p->mp_pgno);
 865                 return EINVAL;
 866         }
 867
 868         return 0;
 869 }
 870
 871 static int
 872 mdbenv_read_meta(MDB_env *env, int *which)
 873 {
 874         MDB_page        *mp0, *mp1;
 875         MDB_meta        *meta[2];
 876         int toggle = 0, rc;
 877
 878         assert(env != NULL);
 879
 880         if ((mp0 = mdbenv_get_page(env, 0)) == NULL ||
 881                 (mp1 = mdbenv_get_page(env, 1)) == NULL)
 882                 return EIO;
 883
 884         rc = mdb_check_meta_page(mp0);
 885         if (rc) return rc;
 886
 887         rc = mdb_check_meta_page(mp1);
 888         if (rc) return rc;
 889
 890         meta[0] = METADATA(mp0);
 891         meta[1] = METADATA(mp1);
 892
 893         if (meta[0]->mm_txnid < meta[1]->mm_txnid)
 894                 toggle = 1;
 895
 896         if (meta[toggle]->mm_txnid > env->me_meta.mm_txnid) {
 897                 bcopy(meta[toggle], &env->me_meta, sizeof(env->me_meta));
 898                 if (which)
 899                         *which = toggle;
 900         }
 901
 902         DPRINTF("Using meta page %d", toggle);
 903
 904         return MDB_SUCCESS;
 905 }
 906
 907 int
 908 mdbenv_create(MDB_env **env)
 909 {
 910         MDB_env *e;
 911
 912         e = calloc(1, sizeof(*e));
 913         if (!e) return ENOMEM;
 914
 915         e->me_meta.mm_mapsize = DEFAULT_MAPSIZE;
 916         e->me_maxreaders = DEFAULT_READERS;
 917         e->me_db.md_env = e;
 918         e->me_fd = -1;
 919         e->me_lfd = -1;
 920         *env = e;
 921         return MDB_SUCCESS;
 922 }
 923
 924 int
 925 mdbenv_set_mapsize(MDB_env *env, size_t size)
 926 {
 927         if (env->me_map)
 928                 return EINVAL;
 929         env->me_mapsize = env->me_meta.mm_mapsize = size;
 930         return MDB_SUCCESS;
 931 }
 932
 933 int
 934 mdbenv_set_maxreaders(MDB_env *env, int readers)
 935 {
 936         env->me_maxreaders = readers;
 937         return MDB_SUCCESS;
 938 }
 939
 940 int
 941 mdbenv_get_maxreaders(MDB_env *env, int *readers)
 942 {
 943         if (!env || !readers)
 944                 return EINVAL;
 945         *readers = env->me_maxreaders;
 946         return MDB_SUCCESS;
 947 }
 948
 949 int
 950 mdbenv_open2(MDB_env *env, unsigned int flags)
 951 {
 952         int i, newenv = 0;
 953
 954         env->me_flags = flags;
 955
 956         if ((i = mdbenv_read_header(env)) != 0) {
 957                 if (i != ENOENT)
 958                         return i;
 959                 DPRINTF("new mdbenv");
 960                 newenv = 1;
 961         }
 962
 963         if (!env->me_mapsize)
 964                 env->me_mapsize = env->me_meta.mm_mapsize;
 965
 966         i = MAP_SHARED;
 967         if (env->me_meta.mm_address && (flags & MDB_FIXEDMAP))
 968                 i |= MAP_FIXED;
 969         env->me_map = mmap(env->me_meta.mm_address, env->me_mapsize, PROT_READ, i,
 970                 env->me_fd, 0);
 971         if (env->me_map == MAP_FAILED)
 972                 return errno;
 973
 974         if (newenv) {
 975                 env->me_meta.mm_mapsize = env->me_mapsize;
 976                 if (flags & MDB_FIXEDMAP)
 977                         env->me_meta.mm_address = env->me_map;
 978                 i = mdbenv_init_meta(env);
 979                 if (i != MDB_SUCCESS) {
 980                         munmap(env->me_map, env->me_mapsize);
 981                         return i;
 982                 }
 983         }
 984
 985         if ((i = mdbenv_read_meta(env, NULL)) != 0)
 986                 return i;
 987
 988         DPRINTF("opened database version %u, pagesize %u",
 989             env->me_meta.mm_version, env->me_meta.mm_stat.ms_psize);
 990         DPRINTF("depth: %u", env->me_meta.mm_stat.ms_depth);
 991         DPRINTF("entries: %lu", env->me_meta.mm_stat.ms_entries);
 992         DPRINTF("branch pages: %lu", env->me_meta.mm_stat.ms_branch_pages);
 993         DPRINTF("leaf pages: %lu", env->me_meta.mm_stat.ms_leaf_pages);
 994         DPRINTF("overflow pages: %lu", env->me_meta.mm_stat.ms_overflow_pages);
 995         DPRINTF("root: %lu", env->me_meta.mm_root);
 996
 997         return MDB_SUCCESS;
 998 }
 999
1000 static void
1001 mdbenv_reader_dest(void *ptr)
1002 {
1003         MDB_reader *reader = ptr;
1004
1005         reader->mr_txnid = 0;
1006         reader->mr_pid = 0;
1007         reader->mr_tid = 0;
1008 }
1009
1010 static void
1011 mdbenv_share_locks(MDB_env *env)
1012 {
1013         struct flock lock_info;
1014
1015         env->me_txns->mt_txnid = env->me_meta.mm_txnid;
1016
1017         memset((void *)&lock_info, 0, sizeof(lock_info));
1018         lock_info.l_type = F_RDLCK;
1019         lock_info.l_whence = SEEK_SET;
1020         lock_info.l_start = 0;
1021         lock_info.l_len = 1;
1022         fcntl(env->me_lfd, F_SETLK, &lock_info);
1023 }
1024
1025 static int
1026 mdbenv_setup_locks(MDB_env *env, char *lpath, int mode, int *excl)
1027 {
1028         int rc;
1029         off_t size, rsize;
1030         struct flock lock_info;
1031
1032         *excl = 0;
1033
1034         if ((env->me_lfd = open(lpath, O_RDWR|O_CREAT, mode)) == -1) {
1035                 rc = errno;
1036                 return rc;
1037         }
1038         /* Try to get exclusive lock. If we succeed, then
1039          * nobody is using the lock region and we should initialize it.
1040          */
1041         memset((void *)&lock_info, 0, sizeof(lock_info));
1042         lock_info.l_type = F_WRLCK;
1043         lock_info.l_whence = SEEK_SET;
1044         lock_info.l_start = 0;
1045         lock_info.l_len = 1;
1046         rc = fcntl(env->me_lfd, F_SETLK, &lock_info);
1047         if (rc == 0) {
1048                 *excl = 1;
1049         } else {
1050                 lock_info.l_type = F_RDLCK;
1051                 rc = fcntl(env->me_lfd, F_SETLK, &lock_info);
1052                 if (rc) {
1053                         rc = errno;
1054                         goto fail;
1055                 }
1056         }
1057         size = lseek(env->me_lfd, 0, SEEK_END);
1058         rsize = (env->me_maxreaders-1) * sizeof(MDB_reader) + sizeof(MDB_txninfo);
1059         if (size < rsize && *excl) {
1060                 if (ftruncate(env->me_lfd, rsize) != 0) {
1061                         rc = errno;
1062                         goto fail;
1063                 }
1064         } else {
1065                 rsize = size;
1066                 size = rsize - sizeof(MDB_txninfo);
1067                 env->me_maxreaders = size/sizeof(MDB_reader) + 1;
1068         }
1069         env->me_txns = mmap(0, rsize, PROT_READ|PROT_WRITE, MAP_SHARED,
1070                 env->me_lfd, 0);
1071         if (env->me_txns == MAP_FAILED) {
1072                 rc = errno;
1073                 goto fail;
1074         }
1075         if (*excl) {
1076                 pthread_mutexattr_t mattr;
1077
1078                 pthread_mutexattr_init(&mattr);
1079                 pthread_mutexattr_setpshared(&mattr, PTHREAD_PROCESS_SHARED);
1080                 pthread_mutex_init(&env->me_txns->mt_mutex, &mattr);
1081                 pthread_mutex_init(&env->me_txns->mt_wmutex, &mattr);
1082                 env->me_txns->mt_version = MDB_VERSION;
1083                 env->me_txns->mt_magic = MDB_MAGIC;
1084                 env->me_txns->mt_txnid = 0;
1085                 env->me_txns->mt_numreaders = 0;
1086
1087         } else {
1088                 if (env->me_txns->mt_magic != MDB_MAGIC) {
1089                         DPRINTF("lock region has invalid magic");
1090                         errno = EINVAL;
1091                 }
1092                 if (env->me_txns->mt_version != MDB_VERSION) {
1093                         DPRINTF("lock region is version %u, expected version %u",
1094                                 env->me_txns->mt_version, MDB_VERSION);
1095                         errno = EINVAL;
1096                 }
1097                 if (errno != EACCES && errno != EAGAIN) {
1098                         rc = errno;
1099                         goto fail;
1100                 }
1101         }
1102         return MDB_SUCCESS;
1103
1104 fail:
1105         close(env->me_lfd);
1106         return rc;
1107
1108 }
1109
1110 int
1111 mdbenv_open(MDB_env *env, const char *path, unsigned int flags, mode_t mode)
1112 {
1113         int             oflags, rc, len, excl;
1114         char *lpath, *dpath;
1115
1116         len = strlen(path);
1117         lpath = malloc(len + sizeof("/lock.mdb") + len + sizeof("/data.db"));
1118         if (!lpath)
1119                 return ENOMEM;
1120         dpath = lpath + len + sizeof("/lock.mdb");
1121         sprintf(lpath, "%s/lock.mdb", path);
1122         sprintf(dpath, "%s/data.mdb", path);
1123
1124         rc = mdbenv_setup_locks(env, lpath, mode, &excl);
1125         if (rc)
1126                 goto leave;
1127
1128         if (F_ISSET(flags, MDB_RDONLY))
1129                 oflags = O_RDONLY;
1130         else
1131                 oflags = O_RDWR | O_CREAT;
1132
1133         if ((env->me_fd = open(dpath, oflags, mode)) == -1)
1134                 return errno;
1135
1136         if ((rc = mdbenv_open2(env, flags)) != MDB_SUCCESS) {
1137                 close(env->me_fd);
1138                 env->me_fd = -1;
1139         } else {
1140                 env->me_path = strdup(path);
1141                 DPRINTF("opened dbenv %p", (void *) env);
1142                 pthread_key_create(&env->me_txkey, mdbenv_reader_dest);
1143                 if (excl)
1144                         mdbenv_share_locks(env);
1145         }
1146
1147
1148 leave:
1149         free(lpath);
1150         return rc;
1151 }
1152
1153 void
1154 mdbenv_close(MDB_env *env)
1155 {
1156         if (env == NULL)
1157                 return;
1158
1159         free(env->me_path);
1160
1161         if (env->me_map) {
1162                 munmap(env->me_map, env->me_mapsize);
1163         }
1164         close(env->me_fd);
1165         if (env->me_txns) {
1166                 size_t size = (env->me_maxreaders-1) * sizeof(MDB_reader) + sizeof(MDB_txninfo);
1167                 munmap(env->me_txns, size);
1168         }
1169         close(env->me_lfd);
1170         free(env);
1171 }
1172
1173 /* Search for key within a leaf page, using binary search.
1174  * Returns the smallest entry larger or equal to the key.
1175  * If exactp is non-null, stores whether the found entry was an exact match
1176  * in *exactp (1 or 0).
1177  * If kip is non-null, stores the index of the found entry in *kip.
1178  * If no entry larger or equal to the key is found, returns NULL.
1179  */
1180 static MDB_node *
1181 mdb_search_node(MDB_db *bt, MDB_page *mp, MDB_val *key,
1182     int *exactp, unsigned int *kip)
1183 {
1184         unsigned int     i = 0;
1185         int              low, high;
1186         int              rc = 0;
1187         MDB_node        *node;
1188         MDB_val  nodekey;
1189
1190         DPRINTF("searching %u keys in %s page %lu",
1191             NUMKEYS(mp),
1192             IS_LEAF(mp) ? "leaf" : "branch",
1193             mp->mp_pgno);
1194
1195         assert(NUMKEYS(mp) > 0);
1196
1197         bzero(&nodekey, sizeof(nodekey));
1198
1199         low = IS_LEAF(mp) ? 0 : 1;
1200         high = NUMKEYS(mp) - 1;
1201         while (low <= high) {
1202                 i = (low + high) >> 1;
1203                 node = NODEPTR(mp, i);
1204
1205                 nodekey.mv_size = node->mn_ksize;
1206                 nodekey.mv_data = NODEKEY(node);
1207
1208                 if (bt->md_cmp)
1209                         rc = bt->md_cmp(key, &nodekey);
1210                 else
1211                         rc = _mdb_cmp(bt, key, &nodekey);
1212
1213                 if (IS_LEAF(mp))
1214                         DPRINTF("found leaf index %u [%.*s], rc = %i",
1215                             i, (int)nodekey.mv_size, (char *)nodekey.mv_data, rc);
1216                 else
1217                         DPRINTF("found branch index %u [%.*s -> %lu], rc = %i",
1218                             i, (int)node->mn_ksize, (char *)NODEKEY(node),
1219                             node->mn_pgno, rc);
1220
1221                 if (rc == 0)
1222                         break;
1223                 if (rc > 0)
1224                         low = i + 1;
1225                 else
1226                         high = i - 1;
1227         }
1228
1229         if (rc > 0) {   /* Found entry is less than the key. */
1230                 i++;    /* Skip to get the smallest entry larger than key. */
1231                 if (i >= NUMKEYS(mp))
1232                         /* There is no entry larger or equal to the key. */
1233                         return NULL;
1234         }
1235         if (exactp)
1236                 *exactp = (rc == 0);
1237         if (kip)        /* Store the key index if requested. */
1238                 *kip = i;
1239
1240         return NODEPTR(mp, i);
1241 }
1242
1243 static void
1244 cursor_pop_page(MDB_cursor *cursor)
1245 {
1246         MDB_ppage       *top;
1247
1248         top = CURSOR_TOP(cursor);
1249         CURSOR_POP(cursor);
1250
1251         DPRINTF("popped page %lu off cursor %p", top->mp_page->mp_pgno, (void *) cursor);
1252
1253         free(top);
1254 }
1255
1256 static MDB_ppage *
1257 cursor_push_page(MDB_cursor *cursor, MDB_page *mp)
1258 {
1259         MDB_ppage       *ppage;
1260
1261         DPRINTF("pushing page %lu on cursor %p", mp->mp_pgno, (void *) cursor);
1262
1263         if ((ppage = calloc(1, sizeof(*ppage))) == NULL)
1264                 return NULL;
1265         ppage->mp_page = mp;
1266         CURSOR_PUSH(cursor, ppage);
1267         return ppage;
1268 }
1269
1270 static MDB_page *
1271 mdbenv_get_page(MDB_env *env, pgno_t pgno)
1272 {
1273         MDB_page *p = NULL;
1274         MDB_txn *txn = env->me_txn;
1275         int found = 0;
1276
1277         if (txn && !SIMPLEQ_EMPTY(txn->mt_u.dirty_queue)) {
1278                 MDB_dpage *dp;
1279                 SIMPLEQ_FOREACH(dp, txn->mt_u.dirty_queue, h.md_next) {
1280                         if (dp->p.mp_pgno == pgno) {
1281                                 p = &dp->p;
1282                                 found = 1;
1283                                 break;
1284                         }
1285                 }
1286         }
1287         if (!found) {
1288                 p = (MDB_page *)(env->me_map + env->me_meta.mm_stat.ms_psize * pgno);
1289         }
1290         return p;
1291 }
1292
1293 static int
1294 mdb_search_page_root(MDB_db *bt, MDB_val *key,
1295     MDB_cursor *cursor, int modify, MDB_pageparent *mpp)
1296 {
1297         MDB_page        *mp = mpp->mp_page;
1298         int rc;
1299
1300         if (cursor && cursor_push_page(cursor, mp) == NULL)
1301                 return MDB_FAIL;
1302
1303         while (IS_BRANCH(mp)) {
1304                 unsigned int     i = 0;
1305                 MDB_node        *node;
1306
1307                 DPRINTF("branch page %lu has %u keys", mp->mp_pgno, NUMKEYS(mp));
1308                 assert(NUMKEYS(mp) > 1);
1309                 DPRINTF("found index 0 to page %lu", NODEPGNO(NODEPTR(mp, 0)));
1310
1311                 if (key == NULL)        /* Initialize cursor to first page. */
1312                         i = 0;
1313                 else {
1314                         int      exact;
1315                         node = mdb_search_node(bt, mp, key, &exact, &i);
1316                         if (node == NULL)
1317                                 i = NUMKEYS(mp) - 1;
1318                         else if (!exact) {
1319                                 assert(i > 0);
1320                                 i--;
1321                         }
1322                 }
1323
1324                 if (key)
1325                         DPRINTF("following index %u for key %.*s",
1326                             i, (int)key->mv_size, (char *)key->mv_data);
1327                 assert(i < NUMKEYS(mp));
1328                 node = NODEPTR(mp, i);
1329
1330                 if (cursor)
1331                         CURSOR_TOP(cursor)->mp_ki = i;
1332
1333                 mpp->mp_parent = mp;
1334                 if ((mp = mdbenv_get_page(bt->md_env, NODEPGNO(node))) == NULL)
1335                         return MDB_FAIL;
1336                 mpp->mp_pi = i;
1337                 mpp->mp_page = mp;
1338
1339                 if (cursor && cursor_push_page(cursor, mp) == NULL)
1340                         return MDB_FAIL;
1341
1342                 if (modify) {
1343                         MDB_dhead *dh = ((MDB_dhead *)mp)-1;
1344                         if ((rc = mdb_touch(bt->md_env->me_txn, mpp)) != 0)
1345                                 return rc;
1346                         dh = ((MDB_dhead *)mpp->mp_page)-1;
1347                         dh->md_parent = mpp->mp_parent;
1348                         dh->md_pi = mpp->mp_pi;
1349                 }
1350
1351                 mp = mpp->mp_page;
1352         }
1353
1354         if (!IS_LEAF(mp)) {
1355                 DPRINTF("internal error, index points to a %02X page!?",
1356                     mp->mp_flags);
1357                 return MDB_FAIL;
1358         }
1359
1360         DPRINTF("found leaf page %lu for key %.*s", mp->mp_pgno,
1361             key ? (int)key->mv_size : 0, key ? (char *)key->mv_data : NULL);
1362
1363         return MDB_SUCCESS;
1364 }
1365
1366 /* Search for the page a given key should be in.
1367  * Stores a pointer to the found page in *mpp.
1368  * If key is NULL, search for the lowest page (used by mdb_cursor_first).
1369  * If cursor is non-null, pushes parent pages on the cursor stack.
1370  * If modify is true, visited pages are updated with new page numbers.
1371  */
1372 static int
1373 mdb_search_page(MDB_db *db, MDB_txn *txn, MDB_val *key,
1374     MDB_cursor *cursor, int modify, MDB_pageparent *mpp)
1375 {
1376         int              rc;
1377         pgno_t           root;
1378
1379         /* Can't modify pages outside a transaction. */
1380         if (txn == NULL && modify) {
1381                 return EINVAL;
1382         }
1383
1384         /* Choose which root page to start with. If a transaction is given
1385          * use the root page from the transaction, otherwise read the last
1386          * committed root page.
1387          */
1388         if (txn == NULL) {
1389                 if ((rc = mdbenv_read_meta(db->md_env, NULL)) != MDB_SUCCESS)
1390                         return rc;
1391                 root = db->md_env->me_meta.mm_root;
1392         } else if (F_ISSET(txn->mt_flags, MDB_TXN_ERROR)) {
1393                 DPRINTF("transaction has failed, must abort");
1394                 return EINVAL;
1395         } else
1396                 root = txn->mt_root;
1397
1398         if (root == P_INVALID) {                /* Tree is empty. */
1399                 DPRINTF("tree is empty");
1400                 return ENOENT;
1401         }
1402
1403         if ((mpp->mp_page = mdbenv_get_page(db->md_env, root)) == NULL)
1404                 return MDB_FAIL;
1405
1406         DPRINTF("root page has flags 0x%X", mpp->mp_page->mp_flags);
1407
1408         if (modify && !F_ISSET(mpp->mp_page->mp_flags, P_DIRTY)) {
1409                 mpp->mp_parent = NULL;
1410                 mpp->mp_pi = 0;
1411                 if ((rc = mdb_touch(txn, mpp)))
1412                         return rc;
1413                 txn->mt_root = mpp->mp_page->mp_pgno;
1414         }
1415
1416         return mdb_search_page_root(db, key, cursor, modify, mpp);
1417 }
1418
1419 static int
1420 mdb_read_data(MDB_db *db, MDB_page *mp, MDB_node *leaf,
1421     MDB_val *data)
1422 {
1423         MDB_page        *omp;           /* overflow mpage */
1424         size_t           max;
1425         pgno_t           pgno;
1426
1427         bzero(data, sizeof(*data));
1428         max = db->md_env->me_meta.mm_stat.ms_psize - PAGEHDRSZ;
1429
1430         if (!F_ISSET(leaf->mn_flags, F_BIGDATA)) {
1431                 data->mv_size = leaf->mn_dsize;
1432                 data->mv_data = NODEDATA(leaf);
1433                 return MDB_SUCCESS;
1434         }
1435
1436         /* Read overflow data.
1437          */
1438         data->mv_size = leaf->mn_dsize;
1439         bcopy(NODEDATA(leaf), &pgno, sizeof(pgno));
1440         if ((omp = mdbenv_get_page(db->md_env, pgno)) == NULL) {
1441                 DPRINTF("read overflow page %lu failed", pgno);
1442                 return MDB_FAIL;
1443         }
1444         data->mv_data = omp;
1445
1446         return MDB_SUCCESS;
1447 }
1448
1449 int
1450 mdb_get(MDB_db *db, MDB_txn *txn,
1451     MDB_val *key, MDB_val *data)
1452 {
1453         int              rc, exact;
1454         MDB_node        *leaf;
1455         MDB_pageparent mpp;
1456
1457         assert(key);
1458         assert(data);
1459         DPRINTF("===> get key [%.*s]", (int)key->mv_size, (char *)key->mv_data);
1460
1461         if (db == NULL)
1462                 return EINVAL;
1463
1464         if (txn != NULL && db->md_env != txn->mt_env) {
1465                 return EINVAL;
1466         }
1467
1468         if (key->mv_size == 0 || key->mv_size > MAXKEYSIZE) {
1469                 return EINVAL;
1470         }
1471
1472         if ((rc = mdb_search_page(db, txn, key, NULL, 0, &mpp)) != MDB_SUCCESS)
1473                 return rc;
1474
1475         leaf = mdb_search_node(db, mpp.mp_page, key, &exact, NULL);
1476         if (leaf && exact)
1477                 rc = mdb_read_data(db, mpp.mp_page, leaf, data);
1478         else {
1479                 rc = ENOENT;
1480         }
1481
1482         return rc;
1483 }
1484
1485 static int
1486 mdb_sibling(MDB_cursor *cursor, int move_right)
1487 {
1488         int              rc;
1489         MDB_node        *indx;
1490         MDB_ppage       *parent, *top;
1491         MDB_page        *mp;
1492
1493         top = CURSOR_TOP(cursor);
1494         if ((parent = SLIST_NEXT(top, mp_entry)) == NULL) {
1495                 return ENOENT;          /* root has no siblings */
1496         }
1497
1498         DPRINTF("parent page is page %lu, index %u",
1499             parent->mp_page->mp_pgno, parent->mp_ki);
1500
1501         cursor_pop_page(cursor);
1502         if (move_right ? (parent->mp_ki + 1 >= NUMKEYS(parent->mp_page))
1503                        : (parent->mp_ki == 0)) {
1504                 DPRINTF("no more keys left, moving to %s sibling",
1505                     move_right ? "right" : "left");
1506                 if ((rc = mdb_sibling(cursor, move_right)) != MDB_SUCCESS)
1507                         return rc;
1508                 parent = CURSOR_TOP(cursor);
1509         } else {
1510                 if (move_right)
1511                         parent->mp_ki++;
1512                 else
1513                         parent->mp_ki--;
1514                 DPRINTF("just moving to %s index key %u",
1515                     move_right ? "right" : "left", parent->mp_ki);
1516         }
1517         assert(IS_BRANCH(parent->mp_page));
1518
1519         indx = NODEPTR(parent->mp_page, parent->mp_ki);
1520         if ((mp = mdbenv_get_page(cursor->mc_db->md_env, indx->mn_pgno)) == NULL)
1521                 return MDB_FAIL;
1522 #if 0
1523         mp->parent = parent->mp_page;
1524         mp->parent_index = parent->mp_ki;
1525 #endif
1526
1527         cursor_push_page(cursor, mp);
1528
1529         return MDB_SUCCESS;
1530 }
1531
1532 static int
1533 mdb_set_key(MDB_db *bt, MDB_page *mp, MDB_node *node,
1534     MDB_val *key)
1535 {
1536         if (key == NULL)
1537                 return 0;
1538
1539         key->mv_size = node->mn_ksize;
1540         key->mv_data = NODEKEY(node);
1541
1542         return 0;
1543 }
1544
1545 static int
1546 mdb_cursor_next(MDB_cursor *cursor, MDB_val *key, MDB_val *data)
1547 {
1548         MDB_ppage       *top;
1549         MDB_page        *mp;
1550         MDB_node        *leaf;
1551
1552         if (cursor->mc_eof) {
1553                 return ENOENT;
1554         }
1555
1556         assert(cursor->mc_initialized);
1557
1558         top = CURSOR_TOP(cursor);
1559         mp = top->mp_page;
1560
1561         DPRINTF("cursor_next: top page is %lu in cursor %p", mp->mp_pgno, (void *) cursor);
1562
1563         if (top->mp_ki + 1 >= NUMKEYS(mp)) {
1564                 DPRINTF("=====> move to next sibling page");
1565                 if (mdb_sibling(cursor, 1) != MDB_SUCCESS) {
1566                         cursor->mc_eof = 1;
1567                         return ENOENT;
1568                 }
1569                 top = CURSOR_TOP(cursor);
1570                 mp = top->mp_page;
1571                 DPRINTF("next page is %lu, key index %u", mp->mp_pgno, top->mp_ki);
1572         } else
1573                 top->mp_ki++;
1574
1575         DPRINTF("==> cursor points to page %lu with %u keys, key index %u",
1576             mp->mp_pgno, NUMKEYS(mp), top->mp_ki);
1577
1578         assert(IS_LEAF(mp));
1579         leaf = NODEPTR(mp, top->mp_ki);
1580
1581         if (data && mdb_read_data(cursor->mc_db, mp, leaf, data) != MDB_SUCCESS)
1582                 return MDB_FAIL;
1583
1584         return mdb_set_key(cursor->mc_db, mp, leaf, key);
1585 }
1586
1587 static int
1588 mdb_cursor_set(MDB_cursor *cursor, MDB_val *key, MDB_val *data,
1589     int *exactp)
1590 {
1591         int              rc;
1592         MDB_node        *leaf;
1593         MDB_ppage       *top;
1594         MDB_pageparent mpp;
1595
1596         assert(cursor);
1597         assert(key);
1598         assert(key->mv_size > 0);
1599
1600         rc = mdb_search_page(cursor->mc_db, cursor->mc_txn, key, cursor, 0, &mpp);
1601         if (rc != MDB_SUCCESS)
1602                 return rc;
1603         assert(IS_LEAF(mpp.mp_page));
1604
1605         top = CURSOR_TOP(cursor);
1606         leaf = mdb_search_node(cursor->mc_db, mpp.mp_page, key, exactp, &top->mp_ki);
1607         if (exactp != NULL && !*exactp) {
1608                 /* MDB_CURSOR_EXACT specified and not an exact match. */
1609                 return ENOENT;
1610         }
1611
1612         if (leaf == NULL) {
1613                 DPRINTF("===> inexact leaf not found, goto sibling");
1614                 if ((rc = mdb_sibling(cursor, 1)) != MDB_SUCCESS)
1615                         return rc;              /* no entries matched */
1616                 top = CURSOR_TOP(cursor);
1617                 top->mp_ki = 0;
1618                 mpp.mp_page = top->mp_page;
1619                 assert(IS_LEAF(mpp.mp_page));
1620                 leaf = NODEPTR(mpp.mp_page, 0);
1621         }
1622
1623         cursor->mc_initialized = 1;
1624         cursor->mc_eof = 0;
1625
1626         if (data && (rc = mdb_read_data(cursor->mc_db, mpp.mp_page, leaf, data)) != MDB_SUCCESS)
1627                 return rc;
1628
1629         rc = mdb_set_key(cursor->mc_db, mpp.mp_page, leaf, key);
1630         if (rc == MDB_SUCCESS) {
1631                 DPRINTF("==> cursor placed on key %.*s",
1632                         (int)key->mv_size, (char *)key->mv_data);
1633                 ;
1634         }
1635
1636         return rc;
1637 }
1638
1639 static int
1640 mdb_cursor_first(MDB_cursor *cursor, MDB_val *key, MDB_val *data)
1641 {
1642         int              rc;
1643         MDB_pageparent  mpp;
1644         MDB_node        *leaf;
1645
1646         rc = mdb_search_page(cursor->mc_db, cursor->mc_txn, NULL, cursor, 0, &mpp);
1647         if (rc != MDB_SUCCESS)
1648                 return rc;
1649         assert(IS_LEAF(mpp.mp_page));
1650
1651         leaf = NODEPTR(mpp.mp_page, 0);
1652         cursor->mc_initialized = 1;
1653         cursor->mc_eof = 0;
1654
1655         if (data && (rc = mdb_read_data(cursor->mc_db, mpp.mp_page, leaf, data)) != MDB_SUCCESS)
1656                 return rc;
1657
1658         return mdb_set_key(cursor->mc_db, mpp.mp_page, leaf, key);
1659 }
1660
1661 int
1662 mdb_cursor_get(MDB_cursor *cursor, MDB_val *key, MDB_val *data,
1663     MDB_cursor_op op)
1664 {
1665         int              rc;
1666         int              exact = 0;
1667
1668         assert(cursor);
1669
1670         switch (op) {
1671         case MDB_CURSOR:
1672         case MDB_CURSOR_EXACT:
1673                 while (CURSOR_TOP(cursor) != NULL)
1674                         cursor_pop_page(cursor);
1675                 if (key == NULL || key->mv_size == 0 || key->mv_size > MAXKEYSIZE) {
1676                         rc = EINVAL;
1677                 } else if (op == MDB_CURSOR_EXACT)
1678                         rc = mdb_cursor_set(cursor, key, data, &exact);
1679                 else
1680                         rc = mdb_cursor_set(cursor, key, data, NULL);
1681                 break;
1682         case MDB_NEXT:
1683                 if (!cursor->mc_initialized)
1684                         rc = mdb_cursor_first(cursor, key, data);
1685                 else
1686                         rc = mdb_cursor_next(cursor, key, data);
1687                 break;
1688         case MDB_FIRST:
1689                 while (CURSOR_TOP(cursor) != NULL)
1690                         cursor_pop_page(cursor);
1691                 rc = mdb_cursor_first(cursor, key, data);
1692                 break;
1693         default:
1694                 DPRINTF("unhandled/unimplemented cursor operation %u", op);
1695                 rc = EINVAL;
1696                 break;
1697         }
1698
1699         return rc;
1700 }
1701
1702 /* Allocate a page and initialize it
1703  */
1704 static MDB_dpage *
1705 mdbenv_new_page(MDB_env *env, uint32_t flags, int num)
1706 {
1707         MDB_dpage       *dp;
1708
1709         assert(env != NULL);
1710         assert(env->me_txn != NULL);
1711
1712         if ((dp = mdb_newpage(env->me_txn, NULL, 0, num)) == NULL)
1713                 return NULL;
1714         DPRINTF("allocated new mpage %lu, page size %u",
1715             dp->p.mp_pgno, env->me_meta.mm_stat.ms_psize);
1716         dp->p.mp_flags = flags | P_DIRTY;
1717         dp->p.mp_lower = PAGEHDRSZ;
1718         dp->p.mp_upper = env->me_meta.mm_stat.ms_psize;
1719
1720         if (IS_BRANCH(&dp->p))
1721                 env->me_meta.mm_stat.ms_branch_pages++;
1722         else if (IS_LEAF(&dp->p))
1723                 env->me_meta.mm_stat.ms_leaf_pages++;
1724         else if (IS_OVERFLOW(&dp->p)) {
1725                 env->me_meta.mm_stat.ms_overflow_pages += num;
1726                 dp->p.mp_pages = num;
1727         }
1728
1729         return dp;
1730 }
1731
1732 static size_t
1733 mdb_leaf_size(MDB_db *db, MDB_val *key, MDB_val *data)
1734 {
1735         size_t           sz;
1736
1737         sz = LEAFSIZE(key, data);
1738         if (data->mv_size >= db->md_env->me_meta.mm_stat.ms_psize / MDB_MINKEYS) {
1739                 /* put on overflow page */
1740                 sz -= data->mv_size - sizeof(pgno_t);
1741         }
1742
1743         return sz + sizeof(indx_t);
1744 }
1745
1746 static size_t
1747 mdb_branch_size(MDB_db *db, MDB_val *key)
1748 {
1749         size_t           sz;
1750
1751         sz = INDXSIZE(key);
1752         if (sz >= db->md_env->me_meta.mm_stat.ms_psize / MDB_MINKEYS) {
1753                 /* put on overflow page */
1754                 /* not implemented */
1755                 /* sz -= key->size - sizeof(pgno_t); */
1756         }
1757
1758         return sz + sizeof(indx_t);
1759 }
1760
1761 static int
1762 mdb_add_node(MDB_db *db, MDB_page *mp, indx_t indx,
1763     MDB_val *key, MDB_val *data, pgno_t pgno, uint8_t flags)
1764 {
1765         unsigned int     i;
1766         size_t           node_size = NODESIZE;
1767         indx_t           ofs;
1768         MDB_node        *node;
1769         MDB_dpage       *ofp = NULL;            /* overflow page */
1770
1771         assert(mp->mp_upper >= mp->mp_lower);
1772
1773         DPRINTF("add node [%.*s] to %s page %lu at index %i, key size %zu",
1774             key ? (int)key->mv_size : 0, key ? (char *)key->mv_data : NULL,
1775             IS_LEAF(mp) ? "leaf" : "branch",
1776             mp->mp_pgno, indx, key ? key->mv_size : 0);
1777
1778         if (key != NULL)
1779                 node_size += key->mv_size;
1780
1781         if (IS_LEAF(mp)) {
1782                 assert(data);
1783                 if (F_ISSET(flags, F_BIGDATA)) {
1784                         /* Data already on overflow page. */
1785                         node_size += sizeof(pgno_t);
1786                 } else if (data->mv_size >= db->md_env->me_meta.mm_stat.ms_psize / MDB_MINKEYS) {
1787                         int ovpages = OVPAGES(data->mv_size, db->md_env->me_meta.mm_stat.ms_psize);
1788                         /* Put data on overflow page. */
1789                         DPRINTF("data size is %zu, put on overflow page",
1790                             data->mv_size);
1791                         node_size += sizeof(pgno_t);
1792                         if ((ofp = mdbenv_new_page(db->md_env, P_OVERFLOW, ovpages)) == NULL)
1793                                 return MDB_FAIL;
1794                         DPRINTF("allocated overflow page %lu", ofp->p.mp_pgno);
1795                         flags |= F_BIGDATA;
1796                 } else {
1797                         node_size += data->mv_size;
1798                 }
1799         }
1800
1801         if (node_size + sizeof(indx_t) > SIZELEFT(mp)) {
1802                 DPRINTF("not enough room in page %lu, got %u ptrs",
1803                     mp->mp_pgno, NUMKEYS(mp));
1804                 DPRINTF("upper - lower = %u - %u = %u", mp->mp_upper, mp->mp_lower,
1805                     mp->mp_upper - mp->mp_lower);
1806                 DPRINTF("node size = %zu", node_size);
1807                 return ENOSPC;
1808         }
1809
1810         /* Move higher pointers up one slot. */
1811         for (i = NUMKEYS(mp); i > indx; i--)
1812                 mp->mp_ptrs[i] = mp->mp_ptrs[i - 1];
1813
1814         /* Adjust free space offsets. */
1815         ofs = mp->mp_upper - node_size;
1816         assert(ofs >= mp->mp_lower + sizeof(indx_t));
1817         mp->mp_ptrs[indx] = ofs;
1818         mp->mp_upper = ofs;
1819         mp->mp_lower += sizeof(indx_t);
1820
1821         /* Write the node data. */
1822         node = NODEPTR(mp, indx);
1823         node->mn_ksize = (key == NULL) ? 0 : key->mv_size;
1824         node->mn_flags = flags;
1825         if (IS_LEAF(mp))
1826                 node->mn_dsize = data->mv_size;
1827         else
1828                 node->mn_pgno = pgno;
1829
1830         if (key)
1831                 bcopy(key->mv_data, NODEKEY(node), key->mv_size);
1832
1833         if (IS_LEAF(mp)) {
1834                 assert(key);
1835                 if (ofp == NULL) {
1836                         if (F_ISSET(flags, F_BIGDATA))
1837                                 bcopy(data->mv_data, node->mn_data + key->mv_size,
1838                                     sizeof(pgno_t));
1839                         else
1840                                 bcopy(data->mv_data, node->mn_data + key->mv_size,
1841                                     data->mv_size);
1842                 } else {
1843                         bcopy(&ofp->p.mp_pgno, node->mn_data + key->mv_size,
1844                             sizeof(pgno_t));
1845                         bcopy(data->mv_data, METADATA(&ofp->p), data->mv_size);
1846                 }
1847         }
1848
1849         return MDB_SUCCESS;
1850 }
1851
1852 static void
1853 mdb_del_node(MDB_db *db, MDB_page *mp, indx_t indx)
1854 {
1855         unsigned int     sz;
1856         indx_t           i, j, numkeys, ptr;
1857         MDB_node        *node;
1858         char            *base;
1859
1860         DPRINTF("delete node %u on %s page %lu", indx,
1861             IS_LEAF(mp) ? "leaf" : "branch", mp->mp_pgno);
1862         assert(indx < NUMKEYS(mp));
1863
1864         node = NODEPTR(mp, indx);
1865         sz = NODESIZE + node->mn_ksize;
1866         if (IS_LEAF(mp)) {
1867                 if (F_ISSET(node->mn_flags, F_BIGDATA))
1868                         sz += sizeof(pgno_t);
1869                 else
1870                         sz += NODEDSZ(node);
1871         }
1872
1873         ptr = mp->mp_ptrs[indx];
1874         numkeys = NUMKEYS(mp);
1875         for (i = j = 0; i < numkeys; i++) {
1876                 if (i != indx) {
1877                         mp->mp_ptrs[j] = mp->mp_ptrs[i];
1878                         if (mp->mp_ptrs[i] < ptr)
1879                                 mp->mp_ptrs[j] += sz;
1880                         j++;
1881                 }
1882         }
1883
1884         base = (char *)mp + mp->mp_upper;
1885         bcopy(base, base + sz, ptr - mp->mp_upper);
1886
1887         mp->mp_lower -= sizeof(indx_t);
1888         mp->mp_upper += sz;
1889 }
1890
1891 int
1892 mdb_cursor_open(MDB_db *db, MDB_txn *txn, MDB_cursor **ret)
1893 {
1894         MDB_cursor      *cursor;
1895
1896         if (db == NULL || ret == NULL)
1897                 return EINVAL;
1898
1899         if (txn != NULL && db->md_env != txn->mt_env) {
1900                 return EINVAL;
1901         }
1902
1903         if ((cursor = calloc(1, sizeof(*cursor))) != NULL) {
1904                 SLIST_INIT(&cursor->mc_stack);
1905                 cursor->mc_db = db;
1906                 cursor->mc_txn = txn;
1907         }
1908
1909         *ret = cursor;
1910
1911         return MDB_SUCCESS;
1912 }
1913
1914 void
1915 mdb_cursor_close(MDB_cursor *cursor)
1916 {
1917         if (cursor != NULL) {
1918                 while (!CURSOR_EMPTY(cursor))
1919                         cursor_pop_page(cursor);
1920
1921 /*              btree_close(cursor->bt); */
1922                 free(cursor);
1923         }
1924 }
1925
1926 static int
1927 mdb_update_key(MDB_db *db, MDB_page *mp, indx_t indx,
1928     MDB_val *key)
1929 {
1930         indx_t                   ptr, i, numkeys;
1931         int                      delta;
1932         size_t                   len;
1933         MDB_node                *node;
1934         char                    *base;
1935
1936         node = NODEPTR(mp, indx);
1937         ptr = mp->mp_ptrs[indx];
1938         DPRINTF("update key %u (ofs %u) [%.*s] to [%.*s] on page %lu",
1939             indx, ptr,
1940             (int)node->mn_ksize, (char *)NODEKEY(node),
1941             (int)key->mv_size, (char *)key->mv_data,
1942             mp->mp_pgno);
1943
1944         delta = key->mv_size - node->mn_ksize;
1945         if (delta) {
1946                 if (delta > 0 && SIZELEFT(mp) < delta) {
1947                         DPRINTF("OUCH! Not enough room, delta = %d", delta);
1948                         return ENOSPC;
1949                 }
1950
1951                 numkeys = NUMKEYS(mp);
1952                 for (i = 0; i < numkeys; i++) {
1953                         if (mp->mp_ptrs[i] <= ptr)
1954                                 mp->mp_ptrs[i] -= delta;
1955                 }
1956
1957                 base = (char *)mp + mp->mp_upper;
1958                 len = ptr - mp->mp_upper + NODESIZE;
1959                 bcopy(base, base - delta, len);
1960                 mp->mp_upper -= delta;
1961
1962                 node = NODEPTR(mp, indx);
1963                 node->mn_ksize = key->mv_size;
1964         }
1965
1966         bcopy(key->mv_data, NODEKEY(node), key->mv_size);
1967
1968         return MDB_SUCCESS;
1969 }
1970
1971 /* Move a node from src to dst.
1972  */
1973 static int
1974 mdb_move_node(MDB_db *bt, MDB_pageparent *src, indx_t srcindx,
1975     MDB_pageparent *dst, indx_t dstindx)
1976 {
1977         int                      rc;
1978         MDB_node                *srcnode;
1979         MDB_val          key, data;
1980
1981         srcnode = NODEPTR(src->mp_page, srcindx);
1982         DPRINTF("moving %s node %u [%.*s] on page %lu to node %u on page %lu",
1983             IS_LEAF(src->mp_page) ? "leaf" : "branch",
1984             srcindx,
1985             (int)srcnode->mn_ksize, (char *)NODEKEY(srcnode),
1986             src->mp_page->mp_pgno,
1987             dstindx, dst->mp_page->mp_pgno);
1988
1989         /* Mark src and dst as dirty. */
1990         if ((rc = mdb_touch(bt->md_env->me_txn, src)) ||
1991             (rc = mdb_touch(bt->md_env->me_txn, dst)))
1992                 return rc;;
1993
1994         /* Add the node to the destination page.
1995          */
1996         key.mv_size = srcnode->mn_ksize;
1997         key.mv_data = NODEKEY(srcnode);
1998         data.mv_size = NODEDSZ(srcnode);
1999         data.mv_data = NODEDATA(srcnode);
2000         rc = mdb_add_node(bt, dst->mp_page, dstindx, &key, &data, NODEPGNO(srcnode),
2001             srcnode->mn_flags);
2002         if (rc != MDB_SUCCESS)
2003                 return rc;
2004
2005         /* Delete the node from the source page.
2006          */
2007         mdb_del_node(bt, src->mp_page, srcindx);
2008
2009         /* Update the parent separators.
2010          */
2011         if (srcindx == 0 && src->mp_pi != 0) {
2012                 DPRINTF("update separator for source page %lu to [%.*s]",
2013                     src->mp_page->mp_pgno, (int)key.mv_size, (char *)key.mv_data);
2014                 if ((rc = mdb_update_key(bt, src->mp_parent, src->mp_pi,
2015                     &key)) != MDB_SUCCESS)
2016                         return rc;
2017         }
2018
2019         if (srcindx == 0 && IS_BRANCH(src->mp_page)) {
2020                 MDB_val  nullkey;
2021                 nullkey.mv_size = 0;
2022                 assert(mdb_update_key(bt, src->mp_page, 0, &nullkey) == MDB_SUCCESS);
2023         }
2024
2025         if (dstindx == 0 && dst->mp_pi != 0) {
2026                 DPRINTF("update separator for destination page %lu to [%.*s]",
2027                     dst->mp_page->mp_pgno, (int)key.mv_size, (char *)key.mv_data);
2028                 if ((rc = mdb_update_key(bt, dst->mp_parent, dst->mp_pi,
2029                     &key)) != MDB_SUCCESS)
2030                         return rc;
2031         }
2032
2033         if (dstindx == 0 && IS_BRANCH(dst->mp_page)) {
2034                 MDB_val  nullkey;
2035                 nullkey.mv_size = 0;
2036                 assert(mdb_update_key(bt, dst->mp_page, 0, &nullkey) == MDB_SUCCESS);
2037         }
2038
2039         return MDB_SUCCESS;
2040 }
2041
2042 static int
2043 mdb_merge(MDB_db *bt, MDB_pageparent *src, MDB_pageparent *dst)
2044 {
2045         int                      rc;
2046         indx_t                   i;
2047         MDB_node                *srcnode;
2048         MDB_val          key, data;
2049         MDB_pageparent  mpp;
2050         MDB_dhead *dh;
2051
2052         DPRINTF("merging page %lu and %lu", src->mp_page->mp_pgno, dst->mp_page->mp_pgno);
2053
2054         assert(src->mp_parent); /* can't merge root page */
2055         assert(dst->mp_parent);
2056         assert(bt->md_env->me_txn != NULL);
2057
2058         /* Mark src and dst as dirty. */
2059         if ((rc = mdb_touch(bt->md_env->me_txn, src)) ||
2060             (rc = mdb_touch(bt->md_env->me_txn, dst)))
2061                 return rc;
2062
2063         /* Move all nodes from src to dst.
2064          */
2065         for (i = 0; i < NUMKEYS(src->mp_page); i++) {
2066                 srcnode = NODEPTR(src->mp_page, i);
2067
2068                 key.mv_size = srcnode->mn_ksize;
2069                 key.mv_data = NODEKEY(srcnode);
2070                 data.mv_size = NODEDSZ(srcnode);
2071                 data.mv_data = NODEDATA(srcnode);
2072                 rc = mdb_add_node(bt, dst->mp_page, NUMKEYS(dst->mp_page), &key,
2073                     &data, NODEPGNO(srcnode), srcnode->mn_flags);
2074                 if (rc != MDB_SUCCESS)
2075                         return rc;
2076         }
2077
2078         DPRINTF("dst page %lu now has %u keys (%.1f%% filled)",
2079             dst->mp_page->mp_pgno, NUMKEYS(dst->mp_page), (float)PAGEFILL(bt->md_env, dst->mp_page) / 10);
2080
2081         /* Unlink the src page from parent.
2082          */
2083         mdb_del_node(bt, src->mp_parent, src->mp_pi);
2084         if (src->mp_pi == 0) {
2085                 key.mv_size = 0;
2086                 if ((rc = mdb_update_key(bt, src->mp_parent, 0, &key)) != MDB_SUCCESS)
2087                         return rc;
2088         }
2089
2090         if (IS_LEAF(src->mp_page))
2091                 bt->md_env->me_meta.mm_stat.ms_leaf_pages--;
2092         else
2093                 bt->md_env->me_meta.mm_stat.ms_branch_pages--;
2094
2095         mpp.mp_page = src->mp_parent;
2096         dh = (MDB_dhead *)src->mp_parent;
2097         dh--;
2098         mpp.mp_parent = dh->md_parent;
2099         mpp.mp_pi = dh->md_pi;
2100
2101         return mdb_rebalance(bt, &mpp);
2102 }
2103
2104 #define FILL_THRESHOLD   250
2105
2106 static int
2107 mdb_rebalance(MDB_db *db, MDB_pageparent *mpp)
2108 {
2109         MDB_node        *node;
2110         MDB_page        *root;
2111         MDB_pageparent npp;
2112         indx_t           si = 0, di = 0;
2113
2114         assert(db != NULL);
2115         assert(db->md_env->me_txn != NULL);
2116         assert(mpp != NULL);
2117
2118         DPRINTF("rebalancing %s page %lu (has %u keys, %.1f%% full)",
2119             IS_LEAF(mpp->mp_page) ? "leaf" : "branch",
2120             mpp->mp_page->mp_pgno, NUMKEYS(mpp->mp_page), (float)PAGEFILL(db->md_env, mpp->mp_page) / 10);
2121
2122         if (PAGEFILL(db->md_env, mpp->mp_page) >= FILL_THRESHOLD) {
2123                 DPRINTF("no need to rebalance page %lu, above fill threshold",
2124                     mpp->mp_page->mp_pgno);
2125                 return MDB_SUCCESS;
2126         }
2127
2128         if (mpp->mp_parent == NULL) {
2129                 if (NUMKEYS(mpp->mp_page) == 0) {
2130                         DPRINTF("tree is completely empty");
2131                         db->md_env->me_txn->mt_root = P_INVALID;
2132                         db->md_env->me_meta.mm_stat.ms_depth--;
2133                         db->md_env->me_meta.mm_stat.ms_leaf_pages--;
2134                 } else if (IS_BRANCH(mpp->mp_page) && NUMKEYS(mpp->mp_page) == 1) {
2135                         DPRINTF("collapsing root page!");
2136                         db->md_env->me_txn->mt_root = NODEPGNO(NODEPTR(mpp->mp_page, 0));
2137                         if ((root = mdbenv_get_page(db->md_env, db->md_env->me_txn->mt_root)) == NULL)
2138                                 return MDB_FAIL;
2139                         db->md_env->me_meta.mm_stat.ms_depth--;
2140                         db->md_env->me_meta.mm_stat.ms_branch_pages--;
2141                 } else
2142                         DPRINTF("root page doesn't need rebalancing");
2143                 return MDB_SUCCESS;
2144         }
2145
2146         /* The parent (branch page) must have at least 2 pointers,
2147          * otherwise the tree is invalid.
2148          */
2149         assert(NUMKEYS(mpp->mp_parent) > 1);
2150
2151         /* Leaf page fill factor is below the threshold.
2152          * Try to move keys from left or right neighbor, or
2153          * merge with a neighbor page.
2154          */
2155
2156         /* Find neighbors.
2157          */
2158         if (mpp->mp_pi == 0) {
2159                 /* We're the leftmost leaf in our parent.
2160                  */
2161                 DPRINTF("reading right neighbor");
2162                 node = NODEPTR(mpp->mp_parent, mpp->mp_pi + 1);
2163                 if ((npp.mp_page = mdbenv_get_page(db->md_env, NODEPGNO(node))) == NULL)
2164                         return MDB_FAIL;
2165                 npp.mp_pi = mpp->mp_pi + 1;
2166                 si = 0;
2167                 di = NUMKEYS(mpp->mp_page);
2168         } else {
2169                 /* There is at least one neighbor to the left.
2170                  */
2171                 DPRINTF("reading left neighbor");
2172                 node = NODEPTR(mpp->mp_parent, mpp->mp_pi - 1);
2173                 if ((npp.mp_page = mdbenv_get_page(db->md_env, NODEPGNO(node))) == NULL)
2174                         return MDB_FAIL;
2175                 npp.mp_pi = mpp->mp_pi - 1;
2176                 si = NUMKEYS(npp.mp_page) - 1;
2177                 di = 0;
2178         }
2179         npp.mp_parent = mpp->mp_parent;
2180
2181         DPRINTF("found neighbor page %lu (%u keys, %.1f%% full)",
2182             npp.mp_page->mp_pgno, NUMKEYS(npp.mp_page), (float)PAGEFILL(db->md_env, npp.mp_page) / 10);
2183
2184         /* If the neighbor page is above threshold and has at least two
2185          * keys, move one key from it.
2186          *
2187          * Otherwise we should try to merge them.
2188          */
2189         if (PAGEFILL(db->md_env, npp.mp_page) >= FILL_THRESHOLD && NUMKEYS(npp.mp_page) >= 2)
2190                 return mdb_move_node(db, &npp, si, mpp, di);
2191         else { /* FIXME: if (has_enough_room()) */
2192                 if (mpp->mp_pi == 0)
2193                         return mdb_merge(db, &npp, mpp);
2194                 else
2195                         return mdb_merge(db, mpp, &npp);
2196         }
2197 }
2198
2199 int
2200 mdb_del(MDB_db *bt, MDB_txn *txn,
2201     MDB_val *key, MDB_val *data)
2202 {
2203         int              rc, exact;
2204         unsigned int     ki;
2205         MDB_node        *leaf;
2206         MDB_pageparent  mpp;
2207
2208         DPRINTF("========> delete key %.*s", (int)key->mv_size, (char *)key->mv_data);
2209
2210         assert(key != NULL);
2211
2212         if (bt == NULL || txn == NULL)
2213                 return EINVAL;
2214
2215         if (F_ISSET(txn->mt_flags, MDB_TXN_RDONLY)) {
2216                 return EINVAL;
2217         }
2218
2219         if (bt->md_env->me_txn != txn) {
2220                 return EINVAL;
2221         }
2222
2223         if (key->mv_size == 0 || key->mv_size > MAXKEYSIZE) {
2224                 return EINVAL;
2225         }
2226
2227         if ((rc = mdb_search_page(bt, txn, key, NULL, 1, &mpp)) != MDB_SUCCESS)
2228                 return rc;
2229
2230         leaf = mdb_search_node(bt, mpp.mp_page, key, &exact, &ki);
2231         if (leaf == NULL || !exact) {
2232                 return ENOENT;
2233         }
2234
2235         if (data && (rc = mdb_read_data(bt, NULL, leaf, data)) != MDB_SUCCESS)
2236                 return rc;
2237
2238         mdb_del_node(bt, mpp.mp_page, ki);
2239         /* add overflow pages to free list */
2240         if (F_ISSET(leaf->mn_flags, F_BIGDATA)) {
2241                 int i, ovpages;
2242                 pgno_t pg;
2243
2244                 bcopy(NODEDATA(leaf), &pg, sizeof(pg));
2245                 ovpages = OVPAGES(NODEDSZ(leaf), bt->md_env->me_meta.mm_stat.ms_psize);
2246                 for (i=0; i<ovpages; i++) {
2247                         mdb_idl_insert(txn->mt_free_pgs, pg);
2248                         pg++;
2249                 }
2250         }
2251         bt->md_env->me_meta.mm_stat.ms_entries--;
2252         rc = mdb_rebalance(bt, &mpp);
2253         if (rc != MDB_SUCCESS)
2254                 txn->mt_flags |= MDB_TXN_ERROR;
2255
2256         return rc;
2257 }
2258
2259 /* Split page <*mpp>, and insert <key,(data|newpgno)> in either left or
2260  * right sibling, at index <*newindxp> (as if unsplit). Updates *mpp and
2261  * *newindxp with the actual values after split, ie if *mpp and *newindxp
2262  * refer to a node in the new right sibling page.
2263  */
2264 static int
2265 mdb_split(MDB_db *bt, MDB_page **mpp, unsigned int *newindxp,
2266     MDB_val *newkey, MDB_val *newdata, pgno_t newpgno)
2267 {
2268         uint8_t          flags;
2269         int              rc = MDB_SUCCESS, ins_new = 0;
2270         indx_t           newindx;
2271         pgno_t           pgno = 0;
2272         unsigned int     i, j, split_indx;
2273         MDB_node        *node;
2274         MDB_val  sepkey, rkey, rdata;
2275         MDB_page        *copy;
2276         MDB_dpage       *mdp, *rdp, *pdp;
2277         MDB_dhead *dh;
2278
2279         assert(bt != NULL);
2280         assert(bt->md_env != NULL);
2281
2282         dh = ((MDB_dhead *)*mpp) - 1;
2283         mdp = (MDB_dpage *)dh;
2284         newindx = *newindxp;
2285
2286         DPRINTF("-----> splitting %s page %lu and adding [%.*s] at index %i",
2287             IS_LEAF(&mdp->p) ? "leaf" : "branch", mdp->p.mp_pgno,
2288             (int)newkey->mv_size, (char *)newkey->mv_data, *newindxp);
2289
2290         if (mdp->h.md_parent == NULL) {
2291                 if ((pdp = mdbenv_new_page(bt->md_env, P_BRANCH, 1)) == NULL)
2292                         return MDB_FAIL;
2293                 mdp->h.md_pi = 0;
2294                 mdp->h.md_parent = &pdp->p;
2295                 bt->md_env->me_txn->mt_root = pdp->p.mp_pgno;
2296                 DPRINTF("root split! new root = %lu", pdp->p.mp_pgno);
2297                 bt->md_env->me_meta.mm_stat.ms_depth++;
2298
2299                 /* Add left (implicit) pointer. */
2300                 if (mdb_add_node(bt, &pdp->p, 0, NULL, NULL,
2301                     mdp->p.mp_pgno, 0) != MDB_SUCCESS)
2302                         return MDB_FAIL;
2303         } else {
2304                 DPRINTF("parent branch page is %lu", mdp->h.md_parent->mp_pgno);
2305         }
2306
2307         /* Create a right sibling. */
2308         if ((rdp = mdbenv_new_page(bt->md_env, mdp->p.mp_flags, 1)) == NULL)
2309                 return MDB_FAIL;
2310         rdp->h.md_parent = mdp->h.md_parent;
2311         rdp->h.md_pi = mdp->h.md_pi + 1;
2312         DPRINTF("new right sibling: page %lu", rdp->p.mp_pgno);
2313
2314         /* Move half of the keys to the right sibling. */
2315         if ((copy = malloc(bt->md_env->me_meta.mm_stat.ms_psize)) == NULL)
2316                 return MDB_FAIL;
2317         bcopy(&mdp->p, copy, bt->md_env->me_meta.mm_stat.ms_psize);
2318         bzero(&mdp->p.mp_ptrs, bt->md_env->me_meta.mm_stat.ms_psize - PAGEHDRSZ);
2319         mdp->p.mp_lower = PAGEHDRSZ;
2320         mdp->p.mp_upper = bt->md_env->me_meta.mm_stat.ms_psize;
2321
2322         split_indx = NUMKEYS(copy) / 2 + 1;
2323
2324         /* First find the separating key between the split pages.
2325          */
2326         bzero(&sepkey, sizeof(sepkey));
2327         if (newindx == split_indx) {
2328                 sepkey.mv_size = newkey->mv_size;
2329                 sepkey.mv_data = newkey->mv_data;
2330         } else {
2331                 node = NODEPTR(copy, split_indx);
2332                 sepkey.mv_size = node->mn_ksize;
2333                 sepkey.mv_data = NODEKEY(node);
2334         }
2335
2336         DPRINTF("separator is [%.*s]", (int)sepkey.mv_size, (char *)sepkey.mv_data);
2337
2338         /* Copy separator key to the parent.
2339          */
2340         if (SIZELEFT(rdp->h.md_parent) < mdb_branch_size(bt, &sepkey)) {
2341                 rc = mdb_split(bt, &rdp->h.md_parent, &rdp->h.md_pi,
2342                     &sepkey, NULL, rdp->p.mp_pgno);
2343
2344                 /* Right page might now have changed parent.
2345                  * Check if left page also changed parent.
2346                  */
2347                 if (rdp->h.md_parent != mdp->h.md_parent &&
2348                     mdp->h.md_pi >= NUMKEYS(mdp->h.md_parent)) {
2349                         mdp->h.md_parent = rdp->h.md_parent;
2350                         mdp->h.md_pi = rdp->h.md_pi - 1;
2351                 }
2352         } else {
2353                 rc = mdb_add_node(bt, rdp->h.md_parent, rdp->h.md_pi,
2354                     &sepkey, NULL, rdp->p.mp_pgno, 0);
2355         }
2356         if (rc != MDB_SUCCESS) {
2357                 free(copy);
2358                 return MDB_FAIL;
2359         }
2360
2361         for (i = j = 0; i <= NUMKEYS(copy); j++) {
2362                 if (i < split_indx) {
2363                         /* Re-insert in left sibling. */
2364                         pdp = mdp;
2365                 } else {
2366                         /* Insert in right sibling. */
2367                         if (i == split_indx)
2368                                 /* Reset insert index for right sibling. */
2369                                 j = (i == newindx && ins_new);
2370                         pdp = rdp;
2371                 }
2372
2373                 if (i == newindx && !ins_new) {
2374                         /* Insert the original entry that caused the split. */
2375                         rkey.mv_data = newkey->mv_data;
2376                         rkey.mv_size = newkey->mv_size;
2377                         if (IS_LEAF(&mdp->p)) {
2378                                 rdata.mv_data = newdata->mv_data;
2379                                 rdata.mv_size = newdata->mv_size;
2380                         } else
2381                                 pgno = newpgno;
2382                         flags = 0;
2383
2384                         ins_new = 1;
2385
2386                         /* Update page and index for the new key. */
2387                         *newindxp = j;
2388                         *mpp = &pdp->p;
2389                 } else if (i == NUMKEYS(copy)) {
2390                         break;
2391                 } else {
2392                         node = NODEPTR(copy, i);
2393                         rkey.mv_data = NODEKEY(node);
2394                         rkey.mv_size = node->mn_ksize;
2395                         if (IS_LEAF(&mdp->p)) {
2396                                 rdata.mv_data = NODEDATA(node);
2397                                 rdata.mv_size = node->mn_dsize;
2398                         } else
2399                                 pgno = node->mn_pgno;
2400                         flags = node->mn_flags;
2401
2402                         i++;
2403                 }
2404
2405                 if (!IS_LEAF(&mdp->p) && j == 0) {
2406                         /* First branch index doesn't need key data. */
2407                         rkey.mv_size = 0;
2408                 }
2409
2410                 rc = mdb_add_node(bt, &pdp->p, j, &rkey, &rdata, pgno,flags);
2411         }
2412
2413         free(copy);
2414         return rc;
2415 }
2416
2417 int
2418 mdb_put(MDB_db *bt, MDB_txn *txn,
2419     MDB_val *key, MDB_val *data, unsigned int flags)
2420 {
2421         int              rc = MDB_SUCCESS, exact;
2422         unsigned int     ki;
2423         MDB_node        *leaf;
2424         MDB_pageparent  mpp;
2425
2426         assert(key != NULL);
2427         assert(data != NULL);
2428
2429         if (bt == NULL || txn == NULL)
2430                 return EINVAL;
2431
2432         if (F_ISSET(txn->mt_flags, MDB_TXN_RDONLY)) {
2433                 return EINVAL;
2434         }
2435
2436         if (bt->md_env->me_txn != txn) {
2437                 return EINVAL;
2438         }
2439
2440         if (key->mv_size == 0 || key->mv_size > MAXKEYSIZE) {
2441                 return EINVAL;
2442         }
2443
2444         DPRINTF("==> put key %.*s, size %zu, data size %zu",
2445                 (int)key->mv_size, (char *)key->mv_data, key->mv_size, data->mv_size);
2446
2447         rc = mdb_search_page(bt, txn, key, NULL, 1, &mpp);
2448         if (rc == MDB_SUCCESS) {
2449                 leaf = mdb_search_node(bt, mpp.mp_page, key, &exact, &ki);
2450                 if (leaf && exact) {
2451                         if (F_ISSET(flags, MDB_NOOVERWRITE)) {
2452                                 DPRINTF("duplicate key %.*s",
2453                                     (int)key->mv_size, (char *)key->mv_data);
2454                                 return EEXIST;
2455                         }
2456                         mdb_del_node(bt, mpp.mp_page, ki);
2457                 }
2458                 if (leaf == NULL) {             /* append if not found */
2459                         ki = NUMKEYS(mpp.mp_page);
2460                         DPRINTF("appending key at index %i", ki);
2461                 }
2462         } else if (rc == ENOENT) {
2463                 MDB_dpage *dp;
2464                 /* new file, just write a root leaf page */
2465                 DPRINTF("allocating new root leaf page");
2466                 if ((dp = mdbenv_new_page(bt->md_env, P_LEAF, 1)) == NULL) {
2467                         return ENOMEM;
2468                 }
2469                 mpp.mp_page = &dp->p;
2470                 txn->mt_root = mpp.mp_page->mp_pgno;
2471                 bt->md_env->me_meta.mm_stat.ms_depth++;
2472                 ki = 0;
2473         }
2474         else
2475                 goto done;
2476
2477         assert(IS_LEAF(mpp.mp_page));
2478         DPRINTF("there are %u keys, should insert new key at index %i",
2479                 NUMKEYS(mpp.mp_page), ki);
2480
2481         if (SIZELEFT(mpp.mp_page) < mdb_leaf_size(bt, key, data)) {
2482                 rc = mdb_split(bt, &mpp.mp_page, &ki, key, data, P_INVALID);
2483         } else {
2484                 /* There is room already in this leaf page. */
2485                 rc = mdb_add_node(bt, mpp.mp_page, ki, key, data, 0, 0);
2486         }
2487
2488         if (rc != MDB_SUCCESS)
2489                 txn->mt_flags |= MDB_TXN_ERROR;
2490         else
2491                 bt->md_env->me_meta.mm_stat.ms_entries++;
2492
2493 done:
2494         return rc;
2495 }
2496
2497 int
2498 mdbenv_get_flags(MDB_env *env, unsigned int *arg)
2499 {
2500         if (!env || !arg)
2501                 return EINVAL;
2502
2503         *arg = env->me_flags;
2504         return MDB_SUCCESS;
2505 }
2506
2507 int
2508 mdbenv_get_path(MDB_env *env, const char **arg)
2509 {
2510         if (!env || !arg)
2511                 return EINVAL;
2512
2513         *arg = env->me_path;
2514         return MDB_SUCCESS;
2515 }
2516
2517 int
2518 mdbenv_stat(MDB_env *env, MDB_stat **arg)
2519 {
2520         if (env == NULL || arg == NULL)
2521                 return EINVAL;
2522
2523         *arg = &env->me_meta.mm_stat;
2524
2525         return MDB_SUCCESS;
2526 }
2527
2528 int mdb_open(MDB_env *env, MDB_txn *txn, const char *name, unsigned int flags, MDB_db **db)
2529 {
2530         if (!name) {
2531                 *db = (MDB_db *)&env->me_db;
2532                 return MDB_SUCCESS;
2533         }
2534         return EINVAL;
2535 }
2536
2537 int mdb_stat(MDB_db *db, MDB_stat **arg)
2538 {
2539         if (db == NULL || arg == NULL)
2540                 return EINVAL;
2541
2542         if (!db->md_parent)
2543                 bcopy(&db->md_env->me_meta.mm_stat, &db->md_stat, sizeof(db->md_stat));
2544
2545         *arg = &db->md_stat;
2546
2547         return MDB_SUCCESS;
2548 }
2549
2550 void mdb_close(MDB_db *db)
2551 {
2552 }