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

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