Add another FIXME comment

[openldap] / libraries / libmdb / mdb.c

diff --git a/libraries/libmdb/mdb.c b/libraries/libmdb/mdb.c
index 6d90db24b5e60d6a6bc3fc1724feecc8657ca4d0..489bbde65618e372a2e9df5b28051fa3f4ef6531 100644 (file)
--- a/libraries/libmdb/mdb.c
+++ b/libraries/libmdb/mdb.c
@@ -48,7 +48,6 @@
 #include <time.h>
 #include <unistd.h>
 #include <pthread.h>
-#include <endian.h>
 
 #include "mdb.h"
 
@@ -93,7 +92,11 @@ typedef struct MDB_rxbody {
 } MDB_rxbody;
 
 #ifndef CACHELINE
-#define CACHELINE      64      /* most CPUs. Itanium uses 128 */
+# ifdef __APPLE__
+#  define CACHELINE    128     /* 64 is too small to contain a mutex */
+# else
+#  define CACHELINE    64      /* most CPUs. Itanium uses 128 */
+# endif
 #endif
 
 typedef struct MDB_reader {
@@ -223,6 +226,9 @@ typedef struct MDB_ppage {                                  /* ordered list of pages */
 } MDB_ppage;
 SLIST_HEAD(page_stack, MDB_ppage);
 
+/* FIXME: tree depth is mostly bounded, we should just
+ * use a fixed array and avoid malloc/pointer chasing
+ */
 #define CURSOR_EMPTY(c)                 SLIST_EMPTY(&(c)->mc_stack)
 #define CURSOR_TOP(c)           SLIST_FIRST(&(c)->mc_stack)
 #define CURSOR_POP(c)           SLIST_REMOVE_HEAD(&(c)->mc_stack, mp_entry)
@@ -386,7 +392,7 @@ static int           mdb_cursor_last(MDB_cursor *cursor,
                            MDB_val *key, MDB_val *data);
 
 static void            mdb_xcursor_init0(MDB_txn *txn, MDB_dbi dbi, MDB_xcursor *mx);
-static void            mdb_xcursor_init1(MDB_txn *txn, MDB_dbi dbi, MDB_xcursor *mx, MDB_db *db);
+static void            mdb_xcursor_init1(MDB_txn *txn, MDB_dbi dbi, MDB_xcursor *mx, MDB_node *node);
 static void            mdb_xcursor_fini(MDB_txn *txn, MDB_dbi dbi, MDB_xcursor *mx);
 
 static size_t           mdb_leaf_size(MDB_env *env, MDB_val *key,
@@ -440,7 +446,11 @@ mdb_cmp(MDB_txn *txn, MDB_dbi dbi, const MDB_val *a, const MDB_val *b)
 static int
 _mdb_cmp(MDB_txn *txn, MDB_dbi dbi, const MDB_val *key1, const MDB_val *key2)
 {
-       if (F_ISSET(txn->mt_dbs[dbi].md_flags, MDB_REVERSEKEY))
+       if (txn->mt_dbs[dbi].md_flags & (MDB_REVERSEKEY
+#if __BYTE_ORDER == __LITTLE_ENDIAN
+               |MDB_INTEGERKEY
+#endif
+       ))
                return memnrcmp(key1->mv_data, key1->mv_size, key2->mv_data, key2->mv_size);
        else
                return memncmp((char *)key1->mv_data, key1->mv_size, key2->mv_data, key2->mv_size);
@@ -990,10 +1000,7 @@ mdbenv_init_meta(MDB_env *env, MDB_meta *meta)
        meta->mm_psize = psize;
        meta->mm_last_pg = 1;
        meta->mm_flags = env->me_flags & 0xffff;
-#if __BYTE_ORDER == __LITTLE_ENDIAN
-       /* freeDB keys are pgno_t's, must compare in int order */
-       meta->mm_flags |= MDB_REVERSEKEY;
-#endif
+       meta->mm_flags |= MDB_INTEGERKEY;
        meta->mm_dbs[0].md_root = P_INVALID;
        meta->mm_dbs[1].md_root = P_INVALID;
 
@@ -1029,7 +1036,8 @@ mdbenv_write_meta(MDB_txn *txn)
        assert(txn != NULL);
        assert(txn->mt_env != NULL);
 
-       DPRINTF("writing meta page for root page %lu", txn->mt_dbs[MAIN_DBI].md_root);
+       DPRINTF("writing meta page %d for root page %lu",
+               !F_ISSET(txn->mt_flags, MDB_TXN_METOGGLE), txn->mt_dbs[MAIN_DBI].md_root);
 
        env = txn->mt_env;
 
@@ -1125,7 +1133,7 @@ mdbenv_get_maxreaders(MDB_env *env, int *readers)
        return MDB_SUCCESS;
 }
 
-int
+static int
 mdbenv_open2(MDB_env *env, unsigned int flags)
 {
        int i, newenv = 0;
@@ -1676,7 +1684,7 @@ mdb_get(MDB_txn *txn, MDB_dbi dbi,
                        MDB_xcursor mx;
 
                        mdb_xcursor_init0(txn, dbi, &mx);
-                       mdb_xcursor_init1(txn, dbi, &mx, NODEDATA(leaf));
+                       mdb_xcursor_init1(txn, dbi, &mx, leaf);
                        rc = mdb_search_page(&mx.mx_txn, mx.mx_cursor.mc_dbi, NULL, NULL, 0, &mpp);
                        if (rc != MDB_SUCCESS)
                                return rc;
@@ -1799,7 +1807,7 @@ mdb_cursor_next(MDB_cursor *cursor, MDB_val *key, MDB_val *data, MDB_cursor_op o
                        return rc;
 
                if (cursor->mc_txn->mt_dbs[cursor->mc_dbi].md_flags & MDB_DUPSORT) {
-                       mdb_xcursor_init1(cursor->mc_txn, cursor->mc_dbi, cursor->mc_xcursor, NODEDATA(leaf));
+                       mdb_xcursor_init1(cursor->mc_txn, cursor->mc_dbi, cursor->mc_xcursor, leaf);
                        rc = mdb_cursor_first(&cursor->mc_xcursor->mx_cursor, data, NULL);
                        if (rc != MDB_SUCCESS)
                                return rc;
@@ -1821,7 +1829,7 @@ mdb_cursor_prev(MDB_cursor *cursor, MDB_val *key, MDB_val *data, MDB_cursor_op o
 
        if (cursor->mc_txn->mt_dbs[cursor->mc_dbi].md_flags & MDB_DUPSORT) {
                if (op == MDB_PREV || op == MDB_PREV_DUP) {
-                       rc = mdb_cursor_next(&cursor->mc_xcursor->mx_cursor, data, NULL, MDB_PREV);
+                       rc = mdb_cursor_prev(&cursor->mc_xcursor->mx_cursor, data, NULL, MDB_PREV);
                        if (op != MDB_PREV || rc == MDB_SUCCESS)
                                return rc;
                }
@@ -1857,7 +1865,7 @@ mdb_cursor_prev(MDB_cursor *cursor, MDB_val *key, MDB_val *data, MDB_cursor_op o
                        return rc;
 
                if (cursor->mc_txn->mt_dbs[cursor->mc_dbi].md_flags & MDB_DUPSORT) {
-                       mdb_xcursor_init1(cursor->mc_txn, cursor->mc_dbi, cursor->mc_xcursor, NODEDATA(leaf));
+                       mdb_xcursor_init1(cursor->mc_txn, cursor->mc_dbi, cursor->mc_xcursor, leaf);
                        rc = mdb_cursor_last(&cursor->mc_xcursor->mx_cursor, data, NULL);
                        if (rc != MDB_SUCCESS)
                                return rc;
@@ -1914,7 +1922,7 @@ mdb_cursor_set(MDB_cursor *cursor, MDB_val *key, MDB_val *data,
                        return rc;
 
                if (cursor->mc_txn->mt_dbs[cursor->mc_dbi].md_flags & MDB_DUPSORT) {
-                       mdb_xcursor_init1(cursor->mc_txn, cursor->mc_dbi, cursor->mc_xcursor, NODEDATA(leaf));
+                       mdb_xcursor_init1(cursor->mc_txn, cursor->mc_dbi, cursor->mc_xcursor, leaf);
                        if (op == MDB_SET || op == MDB_SET_RANGE) {
                                rc = mdb_cursor_first(&cursor->mc_xcursor->mx_cursor, data, NULL);
                        } else {
@@ -1968,7 +1976,7 @@ mdb_cursor_first(MDB_cursor *cursor, MDB_val *key, MDB_val *data)
                        return rc;
 
                if (cursor->mc_txn->mt_dbs[cursor->mc_dbi].md_flags & MDB_DUPSORT) {
-                       mdb_xcursor_init1(cursor->mc_txn, cursor->mc_dbi, cursor->mc_xcursor, NODEDATA(leaf));
+                       mdb_xcursor_init1(cursor->mc_txn, cursor->mc_dbi, cursor->mc_xcursor, leaf);
                        rc = mdb_cursor_first(&cursor->mc_xcursor->mx_cursor, data, NULL);
                        if (rc)
                                return rc;
@@ -2009,7 +2017,7 @@ mdb_cursor_last(MDB_cursor *cursor, MDB_val *key, MDB_val *data)
                        return rc;
 
                if (cursor->mc_txn->mt_dbs[cursor->mc_dbi].md_flags & MDB_DUPSORT) {
-                       mdb_xcursor_init1(cursor->mc_txn, cursor->mc_dbi, cursor->mc_xcursor, NODEDATA(leaf));
+                       mdb_xcursor_init1(cursor->mc_txn, cursor->mc_dbi, cursor->mc_xcursor, leaf);
                        rc = mdb_cursor_last(&cursor->mc_xcursor->mx_cursor, data, NULL);
                        if (rc)
                                return rc;
@@ -2290,21 +2298,32 @@ mdb_xcursor_init0(MDB_txn *txn, MDB_dbi dbi, MDB_xcursor *mx)
 }
 
 static void
-mdb_xcursor_init1(MDB_txn *txn, MDB_dbi dbi, MDB_xcursor *mx, MDB_db *db)
+mdb_xcursor_init1(MDB_txn *txn, MDB_dbi dbi, MDB_xcursor *mx, MDB_node *node)
 {
+       MDB_db *db = NODEDATA(node);
+       MDB_dbi dbn;
        mx->mx_dbs[0] = txn->mt_dbs[0];
        mx->mx_dbs[1] = txn->mt_dbs[1];
        if (dbi > 1) {
                mx->mx_dbs[2] = txn->mt_dbs[dbi];
-               mx->mx_dbs[3] = *db;
+               dbn = 3;
        } else {
-               mx->mx_dbs[2] = *db;
+               dbn = 2;
        }
+       mx->mx_dbs[dbn] = *db;
+       mx->mx_dbxs[dbn].md_name.mv_data = NODEKEY(node);
+       mx->mx_dbxs[dbn].md_name.mv_size = node->mn_ksize;
+       mx->mx_txn.mt_next_pgno = txn->mt_next_pgno;
+       mx->mx_txn.mt_oldest = txn->mt_oldest;
+       mx->mx_txn.mt_u = txn->mt_u;
 }
 
 static void
 mdb_xcursor_fini(MDB_txn *txn, MDB_dbi dbi, MDB_xcursor *mx)
 {
+       txn->mt_next_pgno = mx->mx_txn.mt_next_pgno;
+       txn->mt_oldest = mx->mx_txn.mt_oldest;
+       txn->mt_u = mx->mx_txn.mt_u;
        txn->mt_dbs[0] = mx->mx_dbs[0];
        txn->mt_dbs[1] = mx->mx_dbs[1];
        txn->mt_dbxs[0].md_dirty = mx->mx_dbxs[0].md_dirty;
@@ -2345,6 +2364,23 @@ mdb_cursor_open(MDB_txn *txn, MDB_dbi dbi, MDB_cursor **ret)
        return MDB_SUCCESS;
 }
 
+/* Return the count of duplicate data items for the current key */
+int
+mdb_cursor_count(MDB_cursor *mc, unsigned long *countp)
+{
+       if (mc == NULL || countp == NULL)
+               return EINVAL;
+
+       if (!(mc->mc_txn->mt_dbs[mc->mc_dbi].md_flags & MDB_DUPSORT))
+               return EINVAL;
+
+       if (!mc->mc_xcursor->mx_cursor.mc_initialized)
+               return EINVAL;
+
+       *countp = mc->mc_xcursor->mx_txn.mt_dbs[mc->mc_xcursor->mx_cursor.mc_dbi].md_entries;
+       return MDB_SUCCESS;
+}
+
 void
 mdb_cursor_close(MDB_cursor *cursor)
 {
@@ -2699,7 +2735,7 @@ mdb_del(MDB_txn *txn, MDB_dbi dbi,
                MDB_pageparent mp2;
 
                mdb_xcursor_init0(txn, dbi, &mx);
-               mdb_xcursor_init1(txn, dbi, &mx, NODEDATA(leaf));
+               mdb_xcursor_init1(txn, dbi, &mx, leaf);
                if (flags == MDB_DEL_DUP) {
                        rc = mdb_del(&mx.mx_txn, mx.mx_cursor.mc_dbi, data, NULL, 0);
                        mdb_xcursor_fini(txn, dbi, &mx);
@@ -2983,11 +3019,11 @@ mdb_put0(MDB_txn *txn, MDB_dbi dbi,
                rdata = data;
        }
 
-       if (SIZELEFT(mpp.mp_page) < mdb_leaf_size(txn->mt_env, key, data)) {
-               rc = mdb_split(txn, dbi, &mpp.mp_page, &ki, key, data, P_INVALID);
+       if (SIZELEFT(mpp.mp_page) < mdb_leaf_size(txn->mt_env, key, rdata)) {
+               rc = mdb_split(txn, dbi, &mpp.mp_page, &ki, key, rdata, P_INVALID);
        } else {
                /* There is room already in this leaf page. */
-               rc = mdb_add_node(txn, dbi, mpp.mp_page, ki, key, data, 0, 0);
+               rc = mdb_add_node(txn, dbi, mpp.mp_page, ki, key, rdata, 0, 0);
        }
 
        if (rc != MDB_SUCCESS)
@@ -3012,7 +3048,7 @@ mdb_put0(MDB_txn *txn, MDB_dbi dbi,
                        leaf = NODEPTR(mpp.mp_page, ki);
 put_sub:
                        mdb_xcursor_init0(txn, dbi, &mx);
-                       mdb_xcursor_init1(txn, dbi, &mx, NODEDATA(leaf));
+                       mdb_xcursor_init1(txn, dbi, &mx, leaf);
                        xdata.mv_size = 0;
                        xdata.mv_data = "";
                        if (flags == MDB_NODUPDATA)
@@ -3103,6 +3139,8 @@ int mdb_open(MDB_txn *txn, const char *name, unsigned int flags, MDB_dbi *dbi)
        /* main DB? */
        if (!name) {
                *dbi = MAIN_DBI;
+               if (flags & (MDB_DUPSORT|MDB_REVERSEKEY|MDB_INTEGERKEY))
+                       txn->mt_dbs[MAIN_DBI].md_flags |= (flags & (MDB_DUPSORT|MDB_REVERSEKEY|MDB_INTEGERKEY));
                return MDB_SUCCESS;
        }