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

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