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

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