From 0609aa45b33928f5d80f1bc1edb26e17f84ce95a Mon Sep 17 00:00:00 2001 From: Howard Chu Date: Sat, 3 Sep 2011 17:24:39 -0700 Subject: [PATCH] Reimplement basic ops as cursor ops --- libraries/libmdb/Makefile | 3 +- libraries/libmdb/mdb.c | 825 ++++++++++++++++++++++++++++---------- libraries/libmdb/mdb.h | 17 +- libraries/libmdb/midl.h | 8 +- 4 files changed, 625 insertions(+), 228 deletions(-) diff --git a/libraries/libmdb/Makefile b/libraries/libmdb/Makefile index 425296eef5..3973979e87 100644 --- a/libraries/libmdb/Makefile +++ b/libraries/libmdb/Makefile @@ -4,7 +4,7 @@ OPT = -O2 -g CFLAGS = -pthread $(OPT) $(W) $(XCFLAGS) LDLIBS = -PROGS = mdb_stat mtest mtest2 mtest3 +PROGS = mdb_stat mtest mtest2 mtest3 mtest4 mtest5 all: libmdb.a libmdb.so $(PROGS) clean: @@ -25,6 +25,7 @@ mtest: mtest.o libmdb.a mtest2: mtest2.o libmdb.a mtest3: mtest3.o libmdb.a mtest4: mtest4.o libmdb.a +mtest5: mtest5.o libmdb.a mdb.o: mdb.c mdb.h midl.h $(CC) $(CFLAGS) -fPIC $(CPPFLAGS) -c mdb.c diff --git a/libraries/libmdb/mdb.c b/libraries/libmdb/mdb.c index 4eff6ff8ed..60862cd7c5 100644 --- a/libraries/libmdb/mdb.c +++ b/libraries/libmdb/mdb.c @@ -113,7 +113,6 @@ #endif #endif -#define ULONG unsigned long typedef ULONG pgno_t; #ifndef DEBUG @@ -351,13 +350,15 @@ typedef struct MDB_ppage { /* ordered list of pages */ struct MDB_xcursor; struct MDB_cursor { + struct MDB_xcursor *mc_xcursor; MDB_txn *mc_txn; - MDB_ppage mc_stack[CURSOR_STACK]; /* stack of parent pages */ - unsigned int mc_snum; /* number of pushed pages */ MDB_dbi mc_dbi; - short mc_initialized; /* 1 if initialized */ - short mc_eof; /* 1 if end is reached */ - struct MDB_xcursor *mc_xcursor; + unsigned int mc_snum; /* number of pushed pages */ + unsigned int mc_flags; +#define C_INITIALIZED 0x01 +#define C_EOF 0x02 +#define C_XDIRTY 0x04 + MDB_ppage mc_stack[CURSOR_STACK]; /* stack of parent pages */ }; #define METADATA(p) ((void *)((char *)(p) + PAGEHDRSZ)) @@ -483,52 +484,54 @@ static int mdb_add_node(MDB_txn *txn, MDB_dbi dbi, MDB_page *mp, indx_t indx, MDB_val *key, MDB_val *data, pgno_t pgno, uint8_t flags); static void mdb_del_node(MDB_page *mp, indx_t indx, int ksize); -static int mdb_del0(MDB_txn *txn, MDB_dbi dbi, unsigned int ki, +static int mdb_del0(MDB_cursor *mc, unsigned int ki, MDB_pageparent *mpp, MDB_node *leaf); +#if 0 static int mdb_put0(MDB_txn *txn, MDB_dbi dbi, MDB_val *key, MDB_val *data, unsigned int flags); +#endif static int mdb_read_data(MDB_txn *txn, MDB_node *leaf, MDB_val *data); -static int mdb_rebalance(MDB_txn *txn, MDB_dbi dbi, MDB_pageparent *mp); -static int mdb_update_key(MDB_page *mp, indx_t indx, MDB_val *key); -static int mdb_move_node(MDB_txn *txn, MDB_dbi dbi, +static int mdb_rebalance(MDB_txn *txn, MDB_dbi dbi, MDB_pageparent *mp); +static int mdb_update_key(MDB_page *mp, indx_t indx, MDB_val *key); +static int mdb_move_node(MDB_txn *txn, MDB_dbi dbi, MDB_pageparent *src, indx_t srcindx, MDB_pageparent *dst, indx_t dstindx); -static int mdb_merge(MDB_txn *txn, MDB_dbi dbi, MDB_pageparent *src, +static int mdb_merge(MDB_txn *txn, MDB_dbi dbi, MDB_pageparent *src, MDB_pageparent *dst); -static int mdb_split(MDB_txn *txn, MDB_dbi dbi, MDB_page **mpp, +static int mdb_split(MDB_txn *txn, MDB_dbi dbi, MDB_page **mpp, unsigned int *newindxp, MDB_val *newkey, MDB_val *newdata, pgno_t newpgno); static MDB_dpage *mdb_new_page(MDB_txn *txn, MDB_dbi dbi, uint32_t flags, int num); -static void cursor_pop_page(MDB_cursor *cursor); +static void cursor_pop_page(MDB_cursor *cursor); static MDB_ppage *cursor_push_page(MDB_cursor *cursor, MDB_page *mp); -static int mdb_sibling(MDB_cursor *cursor, int move_right); -static int mdb_cursor_next(MDB_cursor *cursor, +static int mdb_sibling(MDB_cursor *cursor, int move_right); +static int mdb_cursor_next(MDB_cursor *cursor, MDB_val *key, MDB_val *data, MDB_cursor_op op); -static int mdb_cursor_prev(MDB_cursor *cursor, +static int mdb_cursor_prev(MDB_cursor *cursor, MDB_val *key, MDB_val *data, MDB_cursor_op op); -static int mdb_cursor_set(MDB_cursor *cursor, +static int mdb_cursor_set(MDB_cursor *cursor, MDB_val *key, MDB_val *data, MDB_cursor_op op, int *exactp); -static int mdb_cursor_first(MDB_cursor *cursor, +static int mdb_cursor_first(MDB_cursor *cursor, MDB_val *key, MDB_val *data); -static int mdb_cursor_last(MDB_cursor *cursor, +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_page *mp, MDB_node *node); -static void mdb_xcursor_fini(MDB_txn *txn, MDB_dbi dbi, MDB_xcursor *mx); +static void mdb_xcursor_init0(MDB_cursor *mc); +static void mdb_xcursor_init1(MDB_cursor *mc, MDB_page *mp, MDB_node *node); +static void mdb_xcursor_init2(MDB_cursor *mc); +static void mdb_xcursor_fini(MDB_cursor *mc); -static size_t mdb_leaf_size(MDB_env *env, MDB_val *key, +static size_t mdb_leaf_size(MDB_env *env, MDB_val *key, MDB_val *data); -static size_t mdb_branch_size(MDB_env *env, MDB_val *key); +static size_t mdb_branch_size(MDB_env *env, MDB_val *key); -static int memncmp(const void *s1, size_t n1, +static int memncmp(const void *s1, size_t n1, const void *s2, size_t n2); -static int memnrcmp(const void *s1, size_t n1, +static int memnrcmp(const void *s1, size_t n1, const void *s2, size_t n2); #ifdef _WIN32 @@ -609,8 +612,12 @@ mdb_dkey(MDB_val *key, char *buf) unsigned int i; if (key->mv_size > MAXKEYSIZE) return "MAXKEYSIZE"; +#if 0 for (i=0; imv_size; i++) ptr += sprintf(ptr, "%02x", *c++); +#else + sprintf(buf, "%.*s", key->mv_size, key->mv_data); +#endif return buf; } #endif @@ -660,13 +667,18 @@ mdb_alloc_page(MDB_txn *txn, MDB_dbi dbi, MDB_page *parent, unsigned int parent_ if (!txn->mt_env->me_pghead && dbi != FREE_DBI && txn->mt_dbs[FREE_DBI].md_root != P_INVALID) { /* See if there's anything in the free DB */ + MDB_cursor mc; MDB_pageparent mpp; MDB_node *leaf; ULONG *kptr, oldest; mpp.mp_parent = NULL; mpp.mp_pi = 0; - mdb_search_page(txn, FREE_DBI, NULL, NULL, 0, &mpp); + mc.mc_txn = txn; + mc.mc_dbi = FREE_DBI; + mc.mc_snum = 0; + mc.mc_flags = 0; + mdb_search_page(txn, FREE_DBI, NULL, &mc, 0, &mpp); leaf = NODEPTR(mpp.mp_page, 0); kptr = (ULONG *)NODEKEY(leaf); @@ -684,6 +696,7 @@ mdb_alloc_page(MDB_txn *txn, MDB_dbi dbi, MDB_page *parent, unsigned int parent_ /* It's usable, grab it. */ MDB_oldpages *mop; + MDB_ppage *top; MDB_val data; pgno_t *idl; @@ -706,11 +719,10 @@ mdb_alloc_page(MDB_txn *txn, MDB_dbi dbi, MDB_page *parent, unsigned int parent_ } #endif /* drop this IDL from the DB */ - mpp.mp_parent = NULL; - mpp.mp_pi = 0; - mdb_search_page(txn, FREE_DBI, NULL, NULL, 1, &mpp); - leaf = NODEPTR(mpp.mp_page, 0); - mdb_del0(txn, FREE_DBI, 0, &mpp, leaf); + top = CURSOR_TOP(&mc); + top->mp_ki = 0; + mc.mc_flags = C_INITIALIZED; + mdb_cursor_del(&mc, 0); } } if (txn->mt_env->me_pghead) { @@ -998,6 +1010,7 @@ mdb_txn_commit(MDB_txn *txn) MDB_dpage *dp; MDB_env *env; pgno_t next; + MDB_cursor mc; assert(txn != NULL); assert(txn->mt_env != NULL); @@ -1027,6 +1040,10 @@ mdb_txn_commit(MDB_txn *txn) DPRINTF("committing txn %lu %p on mdbenv %p, root page %lu", txn->mt_txnid, txn, (void *)env, txn->mt_dbs[MAIN_DBI].md_root); + mc.mc_txn = txn; + mc.mc_dbi = FREE_DBI; + mc.mc_flags = 0; + /* should only be one record now */ if (env->me_pghead) { MDB_pageparent mpp; @@ -1034,7 +1051,8 @@ mdb_txn_commit(MDB_txn *txn) /* make sure first page of freeDB is touched and on freelist */ mpp.mp_parent = NULL; mpp.mp_pi = 0; - mdb_search_page(txn, FREE_DBI, NULL, NULL, 1, &mpp); + mc.mc_snum = 0; + mdb_search_page(txn, FREE_DBI, NULL, &mc, 1, &mpp); } /* save to free list */ if (!MDB_IDL_IS_ZERO(txn->mt_free_pgs)) { @@ -1047,7 +1065,8 @@ mdb_txn_commit(MDB_txn *txn) key.mv_data = NULL; mpp.mp_parent = NULL; mpp.mp_pi = 0; - mdb_search_page(txn, FREE_DBI, &key, NULL, 1, &mpp); + mc.mc_snum = 0; + mdb_search_page(txn, FREE_DBI, &key, &mc, 1, &mpp); #if DEBUG > 1 { @@ -1071,7 +1090,7 @@ mdb_txn_commit(MDB_txn *txn) do { i = txn->mt_free_pgs[0]; data.mv_size = MDB_IDL_SIZEOF(txn->mt_free_pgs); - rc = mdb_put0(txn, FREE_DBI, &key, &data, 0); + rc = mdb_cursor_put(&mc, &key, &data, 0); if (rc) { mdb_txn_abort(txn); return rc; @@ -1088,7 +1107,7 @@ mdb_txn_commit(MDB_txn *txn) key.mv_data = (char *)&mop->mo_txnid; data.mv_size = MDB_IDL_SIZEOF(mop->mo_pages); data.mv_data = mop->mo_pages; - mdb_put0(txn, FREE_DBI, &key, &data, 0); + mdb_cursor_put(&mc, &key, &data, 0); free(env->me_pghead); env->me_pghead = NULL; } @@ -1100,10 +1119,12 @@ mdb_txn_commit(MDB_txn *txn) MDB_val data; data.mv_size = sizeof(MDB_db); + mc.mc_dbi = MAIN_DBI; + mc.mc_flags = 0; for (i = 2; i < txn->mt_numdbs; i++) { if (txn->mt_dbxs[i].md_dirty) { data.mv_data = &txn->mt_dbs[i]; - mdb_put0(txn, MAIN_DBI, &txn->mt_dbxs[i].md_name, &data, 0); + mdb_cursor_put(&mc, &txn->mt_dbxs[i].md_name, &data, 0); } } } @@ -2019,14 +2040,14 @@ mdb_search_node(MDB_txn *txn, MDB_dbi dbi, MDB_page *mp, MDB_val *key, if (rc > 0) { /* Found entry is less than the key. */ i++; /* Skip to get the smallest entry larger than key. */ - if (i >= NUMKEYS(mp)) - /* There is no entry larger or equal to the key. */ - return NULL; } if (exactp) *exactp = (rc == 0); if (kip) /* Store the key index if requested. */ *kip = i; + if (i >= NUMKEYS(mp)) + /* There is no entry larger or equal to the key. */ + return NULL; /* nodeptr is fake for LEAF2 */ return IS_LEAF2(mp) ? NODEPTR(mp, 0) : NODEPTR(mp, i); @@ -2251,9 +2272,9 @@ int mdb_get(MDB_txn *txn, MDB_dbi dbi, MDB_val *key, MDB_val *data) { - int rc, exact; - MDB_node *leaf; - MDB_pageparent mpp; + MDB_cursor mc; + MDB_xcursor mx; + int exact; DKBUF; assert(key); @@ -2267,36 +2288,16 @@ mdb_get(MDB_txn *txn, MDB_dbi dbi, return EINVAL; } - if ((rc = mdb_search_page(txn, dbi, key, NULL, 0, &mpp)) != MDB_SUCCESS) - return rc; - - leaf = mdb_search_node(txn, dbi, mpp.mp_page, key, &exact, NULL); - if (leaf && exact) { - /* Return first duplicate data item */ - if (F_ISSET(leaf->mn_flags, F_DUPDATA)) { - MDB_xcursor mx; - - mdb_xcursor_init0(txn, dbi, &mx); - mdb_xcursor_init1(txn, dbi, &mx, mpp.mp_page, leaf); - rc = mdb_search_page(&mx.mx_txn, mx.mx_cursor.mc_dbi, NULL, NULL, 0, &mpp); - if (rc != MDB_SUCCESS) - return rc; - if (IS_LEAF2(mpp.mp_page)) { - data->mv_size = txn->mt_dbs[dbi].md_pad; - data->mv_data = LEAF2KEY(mpp.mp_page, 0, data->mv_size); - } else { - leaf = NODEPTR(mpp.mp_page, 0); - data->mv_size = NODEKSZ(leaf); - data->mv_data = NODEKEY(leaf); - } - } else { - rc = mdb_read_data(txn, leaf, data); - } + mc.mc_txn = txn; + mc.mc_dbi = dbi; + mc.mc_flags = 0; + if (txn->mt_dbs[dbi].md_flags & MDB_DUPSORT) { + mc.mc_xcursor = &mx; + mdb_xcursor_init0(&mc); } else { - rc = MDB_NOTFOUND; + mc.mc_xcursor = NULL; } - - return rc; + return mdb_cursor_set(&mc, key, data, MDB_SET, &exact); } static int @@ -2354,11 +2355,11 @@ mdb_cursor_next(MDB_cursor *cursor, MDB_val *key, MDB_val *data, MDB_cursor_op o MDB_node *leaf; int rc; - if (cursor->mc_eof) { + if (cursor->mc_flags & C_EOF) { return MDB_NOTFOUND; } - assert(cursor->mc_initialized); + assert(cursor->mc_flags & C_INITIALIZED); top = CURSOR_TOP(cursor); mp = top->mp_page; @@ -2372,7 +2373,7 @@ mdb_cursor_next(MDB_cursor *cursor, MDB_val *key, MDB_val *data, MDB_cursor_op o return rc; } } else { - cursor->mc_xcursor->mx_cursor.mc_initialized = 0; + cursor->mc_xcursor->mx_cursor.mc_flags = 0; if (op == MDB_NEXT_DUP) return MDB_NOTFOUND; } @@ -2383,7 +2384,7 @@ mdb_cursor_next(MDB_cursor *cursor, MDB_val *key, MDB_val *data, MDB_cursor_op o if (top->mp_ki + 1 >= NUMKEYS(mp)) { DPUTS("=====> move to next sibling page"); if (mdb_sibling(cursor, 1) != MDB_SUCCESS) { - cursor->mc_eof = 1; + cursor->mc_flags |= C_EOF; return MDB_NOTFOUND; } top = CURSOR_TOP(cursor); @@ -2405,7 +2406,7 @@ mdb_cursor_next(MDB_cursor *cursor, MDB_val *key, MDB_val *data, MDB_cursor_op o leaf = NODEPTR(mp, top->mp_ki); if (F_ISSET(leaf->mn_flags, F_DUPDATA)) { - mdb_xcursor_init1(cursor->mc_txn, cursor->mc_dbi, cursor->mc_xcursor, mp, leaf); + mdb_xcursor_init1(cursor, mp, leaf); } if (data) { if ((rc = mdb_read_data(cursor->mc_txn, leaf, data) != MDB_SUCCESS)) @@ -2430,7 +2431,7 @@ mdb_cursor_prev(MDB_cursor *cursor, MDB_val *key, MDB_val *data, MDB_cursor_op o MDB_node *leaf; int rc; - assert(cursor->mc_initialized); + assert(cursor->mc_flags & C_INITIALIZED); top = CURSOR_TOP(cursor); mp = top->mp_page; @@ -2443,7 +2444,7 @@ mdb_cursor_prev(MDB_cursor *cursor, MDB_val *key, MDB_val *data, MDB_cursor_op o if (op != MDB_PREV || rc == MDB_SUCCESS) return rc; } else { - cursor->mc_xcursor->mx_cursor.mc_initialized = 0; + cursor->mc_xcursor->mx_cursor.mc_flags = 0; if (op == MDB_PREV_DUP) return MDB_NOTFOUND; } @@ -2455,7 +2456,7 @@ mdb_cursor_prev(MDB_cursor *cursor, MDB_val *key, MDB_val *data, MDB_cursor_op o if (top->mp_ki == 0) { DPUTS("=====> move to prev sibling page"); if (mdb_sibling(cursor, 0) != MDB_SUCCESS) { - cursor->mc_initialized = 0; + cursor->mc_flags &= ~C_INITIALIZED; return MDB_NOTFOUND; } top = CURSOR_TOP(cursor); @@ -2465,7 +2466,7 @@ mdb_cursor_prev(MDB_cursor *cursor, MDB_val *key, MDB_val *data, MDB_cursor_op o } else top->mp_ki--; - cursor->mc_eof = 0; + cursor->mc_flags &= ~C_EOF; DPRINTF("==> cursor points to page %lu with %u keys, key index %u", mp->mp_pgno, NUMKEYS(mp), top->mp_ki); @@ -2480,7 +2481,7 @@ mdb_cursor_prev(MDB_cursor *cursor, MDB_val *key, MDB_val *data, MDB_cursor_op o leaf = NODEPTR(mp, top->mp_ki); if (F_ISSET(leaf->mn_flags, F_DUPDATA)) { - mdb_xcursor_init1(cursor->mc_txn, cursor->mc_dbi, cursor->mc_xcursor, mp, leaf); + mdb_xcursor_init1(cursor, mp, leaf); } if (data) { if ((rc = mdb_read_data(cursor->mc_txn, leaf, data) != MDB_SUCCESS)) @@ -2512,27 +2513,52 @@ mdb_cursor_set(MDB_cursor *cursor, MDB_val *key, MDB_val *data, assert(key->mv_size > 0); /* See if we're already on the right page */ - if (cursor->mc_initialized) { + if (cursor->mc_flags & C_INITIALIZED) { MDB_val nodekey; + top = CURSOR_TOP(cursor); - /* Don't try this for LEAF2 pages. Maybe support that later. */ - if ((top->mp_page->mp_flags & (P_LEAF|P_LEAF2)) == P_LEAF) { + if (top->mp_page->mp_flags & P_LEAF2) { + nodekey.mv_size = cursor->mc_txn->mt_dbs[cursor->mc_dbi].md_pad; + nodekey.mv_data = LEAF2KEY(top->mp_page, 0, nodekey.mv_size); + } else { leaf = NODEPTR(top->mp_page, 0); MDB_SET_KEY(leaf, &nodekey); - rc = mdb_cmp(cursor->mc_txn, cursor->mc_dbi, key, &nodekey); - if (rc >= 0) { - leaf = NODEPTR(top->mp_page, NUMKEYS(top->mp_page)-1); - MDB_SET_KEY(leaf, &nodekey); - rc = mdb_cmp(cursor->mc_txn, cursor->mc_dbi, key, &nodekey); - if (rc <= 0) { - /* we're already on the right page */ - mpp.mp_page = top->mp_page; - rc = 0; - goto set2; + } + rc = mdb_cmp(cursor->mc_txn, cursor->mc_dbi, key, &nodekey); + if (rc == 0) { +set1: + /* we're already on the right page */ + mpp.mp_page = top->mp_page; + rc = 0; + goto set2; + } + if (rc > 0) { + unsigned int i; + if (NUMKEYS(top->mp_page) > 1) { + if (top->mp_page->mp_flags & P_LEAF2) { + nodekey.mv_data = LEAF2KEY(top->mp_page, + NUMKEYS(top->mp_page)-1, nodekey.mv_size); + } else { + leaf = NODEPTR(top->mp_page, NUMKEYS(top->mp_page)-1); + MDB_SET_KEY(leaf, &nodekey); } + rc = mdb_cmp(cursor->mc_txn, cursor->mc_dbi, key, &nodekey); + if (rc <= 0) goto set1; + } + /* If any parents have right-sibs, search. + * Otherwise, there's nothing further. + */ + for (i=0; imc_snum-1; i++) + if (cursor->mc_stack[i].mp_ki < + NUMKEYS(cursor->mc_stack[i].mp_page)-1) + break; + if (i == cursor->mc_snum - 1) { + /* There are no other pages */ + goto set1; } } } + cursor->mc_snum = 0; rc = mdb_search_page(cursor->mc_txn, cursor->mc_dbi, key, cursor, 0, &mpp); @@ -2560,8 +2586,8 @@ set2: leaf = NODEPTR(mpp.mp_page, 0); } - cursor->mc_initialized = 1; - cursor->mc_eof = 0; + cursor->mc_flags |= C_INITIALIZED; + cursor->mc_flags &= ~C_EOF; if (IS_LEAF2(mpp.mp_page)) { key->mv_size = cursor->mc_txn->mt_dbs[cursor->mc_dbi].md_pad; @@ -2570,7 +2596,7 @@ set2: } if (F_ISSET(leaf->mn_flags, F_DUPDATA)) { - mdb_xcursor_init1(cursor->mc_txn, cursor->mc_dbi, cursor->mc_xcursor, mpp.mp_page, leaf); + mdb_xcursor_init1(cursor, mpp.mp_page, leaf); } if (data) { if (F_ISSET(leaf->mn_flags, F_DUPDATA)) { @@ -2627,8 +2653,8 @@ mdb_cursor_first(MDB_cursor *cursor, MDB_val *key, MDB_val *data) assert(IS_LEAF(mpp.mp_page)); leaf = NODEPTR(mpp.mp_page, 0); - cursor->mc_initialized = 1; - cursor->mc_eof = 0; + cursor->mc_flags |= C_INITIALIZED; + cursor->mc_flags &= ~C_EOF; if (IS_LEAF2(mpp.mp_page)) { key->mv_size = cursor->mc_txn->mt_dbs[cursor->mc_dbi].md_pad; @@ -2638,13 +2664,13 @@ mdb_cursor_first(MDB_cursor *cursor, MDB_val *key, MDB_val *data) if (data) { if (F_ISSET(leaf->mn_flags, F_DUPDATA)) { - mdb_xcursor_init1(cursor->mc_txn, cursor->mc_dbi, cursor->mc_xcursor, mpp.mp_page, leaf); + mdb_xcursor_init1(cursor, mpp.mp_page, leaf); rc = mdb_cursor_first(&cursor->mc_xcursor->mx_cursor, data, NULL); if (rc) return rc; } else { if (cursor->mc_xcursor) - cursor->mc_xcursor->mx_cursor.mc_initialized = 0; + cursor->mc_xcursor->mx_cursor.mc_flags = 0; if ((rc = mdb_read_data(cursor->mc_txn, leaf, data)) != MDB_SUCCESS) return rc; } @@ -2673,8 +2699,8 @@ mdb_cursor_last(MDB_cursor *cursor, MDB_val *key, MDB_val *data) assert(IS_LEAF(mpp.mp_page)); leaf = NODEPTR(mpp.mp_page, NUMKEYS(mpp.mp_page)-1); - cursor->mc_initialized = 1; - cursor->mc_eof = 0; + cursor->mc_flags |= C_INITIALIZED; + cursor->mc_flags &= ~C_EOF; top = CURSOR_TOP(cursor); top->mp_ki = NUMKEYS(top->mp_page) - 1; @@ -2687,7 +2713,7 @@ mdb_cursor_last(MDB_cursor *cursor, MDB_val *key, MDB_val *data) if (data) { if (F_ISSET(leaf->mn_flags, F_DUPDATA)) { - mdb_xcursor_init1(cursor->mc_txn, cursor->mc_dbi, cursor->mc_xcursor, mpp.mp_page, leaf); + mdb_xcursor_init1(cursor, mpp.mp_page, leaf); rc = mdb_cursor_last(&cursor->mc_xcursor->mx_cursor, data, NULL); if (rc) return rc; @@ -2730,12 +2756,13 @@ mdb_cursor_get(MDB_cursor *cursor, MDB_val *key, MDB_val *data, case MDB_GET_MULTIPLE: if (data == NULL || !(cursor->mc_txn->mt_dbs[cursor->mc_dbi].md_flags & MDB_DUPFIXED) || - !cursor->mc_initialized) { + !(cursor->mc_flags & C_INITIALIZED)) { rc = EINVAL; break; } rc = MDB_SUCCESS; - if (!cursor->mc_xcursor->mx_cursor.mc_initialized || cursor->mc_xcursor->mx_cursor.mc_eof) + if (!(cursor->mc_xcursor->mx_cursor.mc_flags & C_INITIALIZED) || + (cursor->mc_xcursor->mx_cursor.mc_flags & C_EOF)) break; goto fetchm; case MDB_NEXT_MULTIPLE: @@ -2744,12 +2771,12 @@ mdb_cursor_get(MDB_cursor *cursor, MDB_val *key, MDB_val *data, rc = EINVAL; break; } - if (!cursor->mc_initialized) + if (!(cursor->mc_flags & C_INITIALIZED)) rc = mdb_cursor_first(cursor, key, data); else rc = mdb_cursor_next(cursor, key, data, MDB_NEXT_DUP); if (rc == MDB_SUCCESS) { - if (cursor->mc_xcursor->mx_cursor.mc_initialized) { + if (cursor->mc_xcursor->mx_cursor.mc_flags & C_INITIALIZED) { MDB_ppage *top; fetchm: top = CURSOR_TOP(&cursor->mc_xcursor->mx_cursor); @@ -2765,7 +2792,7 @@ fetchm: case MDB_NEXT: case MDB_NEXT_DUP: case MDB_NEXT_NODUP: - if (!cursor->mc_initialized) + if (!(cursor->mc_flags & C_INITIALIZED)) rc = mdb_cursor_first(cursor, key, data); else rc = mdb_cursor_next(cursor, key, data, op); @@ -2773,7 +2800,7 @@ fetchm: case MDB_PREV: case MDB_PREV_DUP: case MDB_PREV_NODUP: - if (!cursor->mc_initialized || cursor->mc_eof) + if (!(cursor->mc_flags & C_INITIALIZED) || (cursor->mc_flags & C_EOF)) rc = mdb_cursor_last(cursor, key, data); else rc = mdb_cursor_prev(cursor, key, data, op); @@ -2784,8 +2811,8 @@ fetchm: case MDB_FIRST_DUP: if (data == NULL || !(cursor->mc_txn->mt_dbs[cursor->mc_dbi].md_flags & MDB_DUPSORT) || - !cursor->mc_initialized || - !cursor->mc_xcursor->mx_cursor.mc_initialized) { + !(cursor->mc_flags & C_INITIALIZED) || + !(cursor->mc_xcursor->mx_cursor.mc_flags & C_INITIALIZED)) { rc = EINVAL; break; } @@ -2797,8 +2824,8 @@ fetchm: case MDB_LAST_DUP: if (data == NULL || !(cursor->mc_txn->mt_dbs[cursor->mc_dbi].md_flags & MDB_DUPSORT) || - !cursor->mc_initialized || - !cursor->mc_xcursor->mx_cursor.mc_initialized) { + !(cursor->mc_flags & C_INITIALIZED) || + !(cursor->mc_xcursor->mx_cursor.mc_flags & C_INITIALIZED)) { rc = EINVAL; break; } @@ -2813,6 +2840,335 @@ fetchm: return rc; } +static int +mdb_cursor_touch(MDB_cursor *mc) +{ + MDB_pageparent mpp; + MDB_dhead *dh; + unsigned int i; + int rc; + + mpp.mp_parent = NULL; + mpp.mp_pi = 0; + if (mc->mc_dbi > MAIN_DBI && !mc->mc_txn->mt_dbxs[mc->mc_dbi].md_dirty) { + rc = mdb_search_page(mc->mc_txn, MAIN_DBI, &mc->mc_txn->mt_dbxs[mc->mc_dbi].md_name, + NULL, 1, &mpp); + if (rc) return rc; + mc->mc_txn->mt_dbxs[mc->mc_dbi].md_dirty = 1; + } + for(i=0; imc_snum; i++) { + mpp.mp_page = mc->mc_stack[i].mp_page; + if (!F_ISSET(mpp.mp_page->mp_flags, P_DIRTY)) { + rc = mdb_touch(mc->mc_txn, mc->mc_dbi, &mpp); + if (rc) return rc; + mc->mc_stack[i].mp_page = mpp.mp_page; + if (!i) { + mc->mc_txn->mt_dbs[mc->mc_dbi].md_root = mpp.mp_page->mp_pgno; + } + } + dh = ((MDB_dhead *)mpp.mp_page)-1; + dh->md_parent = mpp.mp_parent; + dh->md_pi = mpp.mp_pi; + mpp.mp_parent = mpp.mp_page; + mpp.mp_pi = mc->mc_stack[i].mp_ki; + } + return MDB_SUCCESS; +} + +int +mdb_cursor_put(MDB_cursor *mc, MDB_val *key, MDB_val *data, + unsigned int flags) +{ + MDB_ppage *top; + MDB_node *leaf; + MDB_val xdata, *rdata, dkey; + MDB_db dummy; + char dbuf[PAGESIZE]; + int do_sub = 0; + size_t nsize; + DKBUF; + int rc, rc2; + + if (F_ISSET(mc->mc_txn->mt_flags, MDB_TXN_RDONLY)) + return EACCES; + + DPRINTF("==> put db %u key [%s], size %zu, data size %zu", + mc->mc_dbi, DKEY(key), key->mv_size, data->mv_size); + + dkey.mv_size = 0; + + if (flags == MDB_CURRENT) { + if (!(mc->mc_flags & C_INITIALIZED)) + return EINVAL; + rc = MDB_SUCCESS; + } else if (mc->mc_txn->mt_dbs[mc->mc_dbi].md_root == P_INVALID) { + MDB_dpage *dp; + /* new database, write a root leaf page */ + DPUTS("allocating new root leaf page"); + if ((dp = mdb_new_page(mc->mc_txn, mc->mc_dbi, P_LEAF, 1)) == NULL) { + return ENOMEM; + } + mc->mc_snum = 0; + cursor_push_page(mc, &dp->p); + mc->mc_txn->mt_dbs[mc->mc_dbi].md_root = dp->p.mp_pgno; + mc->mc_txn->mt_dbs[mc->mc_dbi].md_depth++; + mc->mc_txn->mt_dbxs[mc->mc_dbi].md_dirty = 1; + if ((mc->mc_txn->mt_dbs[mc->mc_dbi].md_flags & (MDB_DUPSORT|MDB_DUPFIXED)) + == MDB_DUPFIXED) + dp->p.mp_flags |= P_LEAF2; + mc->mc_flags |= C_INITIALIZED; + rc = MDB_NOTFOUND; + goto top; + } else { + int exact = 0; + rc = mdb_cursor_set(mc, key, NULL, MDB_SET, &exact); + if (flags == MDB_NOOVERWRITE && rc == 0) { + DPRINTF("duplicate key [%s]", DKEY(key)); + return MDB_KEYEXIST; + } + if (rc && rc != MDB_NOTFOUND) + return rc; + } + + /* Cursor is positioned, now make sure all pages are writable */ + rc2 = mdb_cursor_touch(mc); + if (rc2) return rc2; + +top: + top = CURSOR_TOP(mc); + + /* The key already exists */ + if (rc == MDB_SUCCESS) { + /* there's only a key anyway, so this is a no-op */ + if (IS_LEAF2(top->mp_page)) { + int ksize = mc->mc_txn->mt_dbs[mc->mc_dbi].md_pad; + if (key->mv_size != ksize) + return EINVAL; + if (flags == MDB_CURRENT) { + char *ptr = LEAF2KEY(top->mp_page, top->mp_ki, ksize); + memcpy(ptr, key->mv_data, ksize); + } + return MDB_SUCCESS; + } + + leaf = NODEPTR(top->mp_page, top->mp_ki); + + /* DB has dups? */ + if (F_ISSET(mc->mc_txn->mt_dbs[mc->mc_dbi].md_flags, MDB_DUPSORT)) { + /* Was a single item before, must convert now */ + if (!F_ISSET(leaf->mn_flags, F_DUPDATA)) { + dkey.mv_size = NODEDSZ(leaf); + dkey.mv_data = dbuf; + memcpy(dbuf, NODEDATA(leaf), dkey.mv_size); + /* data matches, ignore it */ + if (!mdb_dcmp(mc->mc_txn, mc->mc_dbi, data, &dkey)) + return (flags == MDB_NODUPDATA) ? MDB_KEYEXIST : MDB_SUCCESS; + memset(&dummy, 0, sizeof(dummy)); + if (mc->mc_txn->mt_dbs[mc->mc_dbi].md_flags & MDB_DUPFIXED) { + dummy.md_pad = data->mv_size; + dummy.md_flags = MDB_DUPFIXED; + if (mc->mc_txn->mt_dbs[mc->mc_dbi].md_flags & MDB_INTEGERDUP) + dummy.md_flags |= MDB_INTEGERKEY; + } + dummy.md_root = P_INVALID; + if (dkey.mv_size == sizeof(MDB_db)) { + memcpy(NODEDATA(leaf), &dummy, sizeof(dummy)); + goto put_sub; + } + mdb_del_node(top->mp_page, top->mp_ki, 0); + do_sub = 1; + rdata = &xdata; + xdata.mv_size = sizeof(MDB_db); + xdata.mv_data = &dummy; + goto new_sub; + } + goto put_sub; + } + /* same size, just replace it */ + if (!F_ISSET(leaf->mn_flags, F_BIGDATA) && + NODEDSZ(leaf) == data->mv_size) { + memcpy(NODEDATA(leaf), data->mv_data, data->mv_size); + goto done; + } + mdb_del_node(top->mp_page, top->mp_ki, 0); + } else { + DPRINTF("inserting key at index %i", top->mp_ki); + } + + rdata = data; + +new_sub: + nsize = IS_LEAF2(top->mp_page) ? key->mv_size : mdb_leaf_size(mc->mc_txn->mt_env, key, rdata); + if (SIZELEFT(top->mp_page) < nsize) { + rc = mdb_split(mc->mc_txn, mc->mc_dbi, &top->mp_page, &top->mp_ki, key, rdata, P_INVALID); + } else { + /* There is room already in this leaf page. */ + rc = mdb_add_node(mc->mc_txn, mc->mc_dbi, top->mp_page, top->mp_ki, key, rdata, 0, 0); + } + + if (rc != MDB_SUCCESS) + mc->mc_txn->mt_flags |= MDB_TXN_ERROR; + else { + /* Remember if we just added a subdatabase */ + if (flags & F_SUBDATA) { + leaf = NODEPTR(top->mp_page, top->mp_ki); + leaf->mn_flags |= F_SUBDATA; + } + + /* Now store the actual data in the child DB. Note that we're + * storing the user data in the keys field, so there are strict + * size limits on dupdata. The actual data fields of the child + * DB are all zero size. + */ + if (do_sub) { + leaf = NODEPTR(top->mp_page, top->mp_ki); +put_sub: + if (flags == MDB_CURRENT) + mdb_xcursor_init2(mc); + else + mdb_xcursor_init1(mc, top->mp_page, leaf); + xdata.mv_size = 0; + xdata.mv_data = ""; + if (flags == MDB_NODUPDATA) + flags = MDB_NOOVERWRITE; + /* converted, write the original data first */ + if (dkey.mv_size) { + rc = mdb_cursor_put(&mc->mc_xcursor->mx_cursor, &dkey, &xdata, flags); + if (rc) return rc; + leaf->mn_flags |= F_DUPDATA; + } + rc = mdb_cursor_put(&mc->mc_xcursor->mx_cursor, data, &xdata, flags); + mdb_xcursor_fini(mc); + memcpy(NODEDATA(leaf), + &mc->mc_xcursor->mx_txn.mt_dbs[mc->mc_xcursor->mx_cursor.mc_dbi], + sizeof(MDB_db)); + } + mc->mc_txn->mt_dbs[mc->mc_dbi].md_entries++; + } +done: + return rc; +} + +int +mdb_cursor_del(MDB_cursor *mc, unsigned int flags) +{ + MDB_pageparent mpp; + MDB_ppage *top; + MDB_node *leaf; + int rc; + + if (F_ISSET(mc->mc_txn->mt_flags, MDB_TXN_RDONLY)) + return EACCES; + + if (!mc->mc_flags & C_INITIALIZED) + return EINVAL; + + rc = mdb_cursor_touch(mc); + if (rc) return rc; + + top = CURSOR_TOP(mc); + leaf = NODEPTR(top->mp_page, top->mp_ki); + mpp.mp_page = top->mp_page; + if (mc->mc_snum > 1) { + MDB_ppage *parent = CURSOR_PARENT(mc); + mpp.mp_parent = parent->mp_page; + mpp.mp_pi = parent->mp_ki; + } else { + mpp.mp_parent = NULL; + mpp.mp_pi = 0; + } + + if (!IS_LEAF2(top->mp_page) && F_ISSET(leaf->mn_flags, F_DUPDATA)) { + MDB_pageparent mp2; + + if (flags != MDB_NODUPDATA) { +/* mdb_xcursor_init2(mc); */ + rc = mdb_cursor_del(&mc->mc_xcursor->mx_cursor, 0); + mdb_xcursor_fini(mc); + /* If sub-DB still has entries, we're done */ + if (mc->mc_xcursor->mx_txn.mt_dbs[mc->mc_xcursor->mx_cursor.mc_dbi].md_root + != P_INVALID) { + memcpy(NODEDATA(leaf), + &mc->mc_xcursor->mx_txn.mt_dbs[mc->mc_xcursor->mx_cursor.mc_dbi], + sizeof(MDB_db)); + mc->mc_txn->mt_dbs[mc->mc_dbi].md_entries--; + return rc; + } + /* otherwise fall thru and delete the sub-DB */ + } + + /* add all the child DB's pages to the free list */ + mc->mc_xcursor->mx_cursor.mc_snum = 0; + rc = mdb_search_page(&mc->mc_xcursor->mx_txn, mc->mc_xcursor->mx_cursor.mc_dbi, + NULL, &mc->mc_xcursor->mx_cursor, 0, &mp2); + if (rc == MDB_SUCCESS) { + MDB_ppage *top, *parent; + MDB_dpage *dp; + ID2 mid; + MDB_node *ni; + unsigned int i; + int dirty_root = 0; + + mc->mc_txn->mt_dbs[mc->mc_dbi].md_entries -= + mc->mc_xcursor->mx_txn.mt_dbs[mc->mc_xcursor->mx_cursor.mc_dbi].md_entries; + + cursor_pop_page(&mc->mc_xcursor->mx_cursor); + if (mc->mc_xcursor->mx_cursor.mc_snum) { +#if 0 + if (mc->mc_xcursor->mx_cursor.mc_stack[0].mp_page->mp_flags & P_DIRTY) { + dirty_root = 1; + } +#endif + while (mc->mc_xcursor->mx_cursor.mc_snum > 1) { + top = CURSOR_TOP(&mc->mc_xcursor->mx_cursor); + parent = CURSOR_PARENT(&mc->mc_xcursor->mx_cursor); + for (i=0; imp_page); i++) { + MDB_page *mp; + pgno_t pg; + ni = NODEPTR(top->mp_page, i); + pg = NODEPGNO(ni); + if ((rc = mdb_get_page(mc->mc_txn, pg, &mp))) + return rc; +#if 0 + if (mp->mp_flags & P_DIRTY) { + /* drop it */ + mid.mid = pg; + mdb_mid2l_delete(mc->mc_txn->mt_u.dirty_list, &mid); + dp = mid.mptr; + dp->h.md_parent = (MDB_page *)mc->mc_txn->mt_env->me_dpages; + mc->mc_txn->mt_env->me_dpages = dp; + } else +#endif + { + /* free it */ + mdb_midl_insert(mc->mc_txn->mt_free_pgs, pg); + } + } + rc = mdb_sibling(&mc->mc_xcursor->mx_cursor, 1); + if (rc) break; + } + } +#if 0 + if (dirty_root) { + /* drop it */ + mid.mid = mc->mc_xcursor->mx_txn.mt_dbs[mc->mc_xcursor->mx_cursor.mc_dbi].md_root; + mdb_mid2l_delete(mc->mc_txn->mt_u.dirty_list, &mid); + dp = mid.mptr; + dp->h.md_parent = (MDB_page *)mc->mc_txn->mt_env->me_dpages; + mc->mc_txn->mt_env->me_dpages = dp; + } else +#endif + { + /* free it */ + mdb_midl_insert(mc->mc_txn->mt_free_pgs, + mc->mc_xcursor->mx_txn.mt_dbs[mc->mc_xcursor->mx_cursor.mc_dbi].md_root); + } + } + } + + return mdb_del0(mc, top->mp_ki, &mpp, leaf); +} + /* Allocate a page and initialize it */ static MDB_dpage * @@ -3027,17 +3383,18 @@ mdb_del_node(MDB_page *mp, indx_t indx, int ksize) } static void -mdb_xcursor_init0(MDB_txn *txn, MDB_dbi dbi, MDB_xcursor *mx) +mdb_xcursor_init0(MDB_cursor *mc) { + MDB_xcursor *mx = mc->mc_xcursor; MDB_dbi dbn; - mx->mx_txn = *txn; + mx->mx_txn = *mc->mc_txn; mx->mx_txn.mt_dbxs = mx->mx_dbxs; mx->mx_txn.mt_dbs = mx->mx_dbs; - mx->mx_dbxs[0] = txn->mt_dbxs[0]; - mx->mx_dbxs[1] = txn->mt_dbxs[1]; - if (dbi > 1) { - mx->mx_dbxs[2] = txn->mt_dbxs[dbi]; + mx->mx_dbxs[0] = mc->mc_txn->mt_dbxs[0]; + mx->mx_dbxs[1] = mc->mc_txn->mt_dbxs[1]; + if (mc->mc_dbi > 1) { + mx->mx_dbxs[2] = mc->mc_txn->mt_dbxs[mc->mc_dbi]; dbn = 2; } else { dbn = 1; @@ -3048,48 +3405,67 @@ mdb_xcursor_init0(MDB_txn *txn, MDB_dbi dbi, MDB_xcursor *mx) mx->mx_dbxs[dbn+1].md_dirty = 0; mx->mx_txn.mt_numdbs = dbn+2; - mx->mx_cursor.mc_snum = 0; + mx->mx_cursor.mc_xcursor = NULL; mx->mx_cursor.mc_txn = &mx->mx_txn; mx->mx_cursor.mc_dbi = dbn+1; } static void -mdb_xcursor_init1(MDB_txn *txn, MDB_dbi dbi, MDB_xcursor *mx, MDB_page *mp, MDB_node *node) +mdb_xcursor_init1(MDB_cursor *mc, MDB_page *mp, MDB_node *node) { MDB_db *db = NODEDATA(node); + MDB_xcursor *mx = mc->mc_xcursor; 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[0] = mc->mc_txn->mt_dbs[0]; + mx->mx_dbs[1] = mc->mc_txn->mt_dbs[1]; + if (mc->mc_dbi > 1) { + mx->mx_dbs[2] = mc->mc_txn->mt_dbs[mc->mc_dbi]; dbn = 3; } else { dbn = 2; } - DPRINTF("Sub-db %u for db %u root page %lu", dbn, dbi, db->md_root); + DPRINTF("Sub-db %u for db %u root page %lu", dbn, mc->mc_dbi, db->md_root); mx->mx_dbs[dbn] = *db; if (F_ISSET(mp->mp_flags, P_DIRTY)) mx->mx_dbxs[dbn].md_dirty = 1; 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_u = txn->mt_u; - mx->mx_cursor.mc_initialized = 0; - mx->mx_cursor.mc_eof = 0; + mx->mx_txn.mt_next_pgno = mc->mc_txn->mt_next_pgno; + mx->mx_txn.mt_u = mc->mc_txn->mt_u; + mx->mx_cursor.mc_snum = 0; + mx->mx_cursor.mc_flags = 0; +} + +static void +mdb_xcursor_init2(MDB_cursor *mc) +{ + MDB_xcursor *mx = mc->mc_xcursor; + MDB_dbi dbn; + mx->mx_dbs[0] = mc->mc_txn->mt_dbs[0]; + mx->mx_dbs[1] = mc->mc_txn->mt_dbs[1]; + if (mc->mc_dbi > 1) { + mx->mx_dbs[2] = mc->mc_txn->mt_dbs[mc->mc_dbi]; + dbn = 3; + } else { + dbn = 2; + } + DPRINTF("Sub-db %u for db %u root page %lu", dbn, mc->mc_dbi, + mx->mx_dbs[dbn].md_root); + mx->mx_txn.mt_next_pgno = mc->mc_txn->mt_next_pgno; } static void -mdb_xcursor_fini(MDB_txn *txn, MDB_dbi dbi, MDB_xcursor *mx) +mdb_xcursor_fini(MDB_cursor *mc) { - txn->mt_next_pgno = mx->mx_txn.mt_next_pgno; - 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; - txn->mt_dbxs[1].md_dirty = mx->mx_dbxs[1].md_dirty; - if (dbi > 1) { - txn->mt_dbs[dbi] = mx->mx_dbs[2]; - txn->mt_dbxs[dbi].md_dirty = mx->mx_dbxs[2].md_dirty; + MDB_xcursor *mx = mc->mc_xcursor; + mc->mc_txn->mt_next_pgno = mx->mx_txn.mt_next_pgno; + mc->mc_txn->mt_dbs[0] = mx->mx_dbs[0]; + mc->mc_txn->mt_dbs[1] = mx->mx_dbs[1]; + mc->mc_txn->mt_dbxs[0].md_dirty = mx->mx_dbxs[0].md_dirty; + mc->mc_txn->mt_dbxs[1].md_dirty = mx->mx_dbxs[1].md_dirty; + if (mc->mc_dbi > 1) { + mc->mc_txn->mt_dbs[mc->mc_dbi] = mx->mx_dbs[2]; + mc->mc_txn->mt_dbxs[mc->mc_dbi].md_dirty = mx->mx_dbxs[2].md_dirty; } } @@ -3111,7 +3487,7 @@ mdb_cursor_open(MDB_txn *txn, MDB_dbi dbi, MDB_cursor **ret) if (txn->mt_dbs[dbi].md_flags & MDB_DUPSORT) { MDB_xcursor *mx = (MDB_xcursor *)(cursor + 1); cursor->mc_xcursor = mx; - mdb_xcursor_init0(txn, dbi, mx); + mdb_xcursor_init0(cursor); } } else { return ENOMEM; @@ -3140,7 +3516,7 @@ mdb_cursor_count(MDB_cursor *mc, unsigned long *countp) if (!F_ISSET(leaf->mn_flags, F_DUPDATA)) { *countp = 1; } else { - if (!mc->mc_xcursor->mx_cursor.mc_initialized) + if (!(mc->mc_xcursor->mx_cursor.mc_flags & C_INITIALIZED)) return EINVAL; *countp = mc->mc_xcursor->mx_txn.mt_dbs[mc->mc_xcursor->mx_cursor.mc_dbi].md_entries; @@ -3148,6 +3524,32 @@ mdb_cursor_count(MDB_cursor *mc, unsigned long *countp) return MDB_SUCCESS; } +#if 0 +static void +mdb_cursor_reset(MDB_cursor *mc) +{ + if (mc != NULL) { + if (mc->mc_flags & C_XDIRTY) { + + mdb_xcursor_fini(mc); + + /* If sub-DB still has entries, update root record */ + if (mc->mc_xcursor->mx_txn.mt_dbs[mc->mc_xcursor->mx_cursor.mc_dbi].md_root + != P_INVALID) { + MDB_ppage *top; + MDB_node *leaf; + top = CURSOR_TOP(mc); + leaf = NODEPTR(top->mp_page, top->mp_ki); + memcpy(NODEDATA(leaf), + &mc->mc_xcursor->mx_txn.mt_dbs[mc->mc_xcursor->mx_cursor.mc_dbi], + sizeof(MDB_db)); + } + mc->mc_flags ^= C_XDIRTY; + } + } +} +#endif + void mdb_cursor_close(MDB_cursor *cursor) { @@ -3394,8 +3796,10 @@ mdb_rebalance(MDB_txn *txn, MDB_dbi dbi, MDB_pageparent *mpp) txn->mt_dbs[dbi].md_root = P_INVALID; txn->mt_dbs[dbi].md_depth = 0; txn->mt_dbs[dbi].md_leaf_pages = 0; + mdb_midl_insert(txn->mt_free_pgs, mpp->mp_page->mp_pgno); } else if (IS_BRANCH(mpp->mp_page) && NUMKEYS(mpp->mp_page) == 1) { DPUTS("collapsing root page!"); + mdb_midl_insert(txn->mt_free_pgs, mpp->mp_page->mp_pgno); txn->mt_dbs[dbi].md_root = NODEPGNO(NODEPTR(mpp->mp_page, 0)); if ((rc = mdb_get_page(txn, txn->mt_dbs[dbi].md_root, &root))) return rc; @@ -3460,7 +3864,7 @@ mdb_rebalance(MDB_txn *txn, MDB_dbi dbi, MDB_pageparent *mpp) } static int -mdb_del0(MDB_txn *txn, MDB_dbi dbi, unsigned int ki, MDB_pageparent *mpp, MDB_node *leaf) +mdb_del0(MDB_cursor *mc, unsigned int ki, MDB_pageparent *mpp, MDB_node *leaf) { int rc; @@ -3470,18 +3874,18 @@ mdb_del0(MDB_txn *txn, MDB_dbi dbi, unsigned int ki, MDB_pageparent *mpp, MDB_no pgno_t pg; memcpy(&pg, NODEDATA(leaf), sizeof(pg)); - ovpages = OVPAGES(NODEDSZ(leaf), txn->mt_env->me_psize); + ovpages = OVPAGES(NODEDSZ(leaf), mc->mc_txn->mt_env->me_psize); for (i=0; imt_free_pgs, pg); + mdb_midl_insert(mc->mc_txn->mt_free_pgs, pg); pg++; } } - mdb_del_node(mpp->mp_page, ki, txn->mt_dbs[dbi].md_pad); - txn->mt_dbs[dbi].md_entries--; - rc = mdb_rebalance(txn, dbi, mpp); + mdb_del_node(mpp->mp_page, ki, mc->mc_txn->mt_dbs[mc->mc_dbi].md_pad); + mc->mc_txn->mt_dbs[mc->mc_dbi].md_entries--; + rc = mdb_rebalance(mc->mc_txn, mc->mc_dbi, mpp); if (rc != MDB_SUCCESS) - txn->mt_flags |= MDB_TXN_ERROR; + mc->mc_txn->mt_flags |= MDB_TXN_ERROR; return rc; } @@ -3490,10 +3894,11 @@ int mdb_del(MDB_txn *txn, MDB_dbi dbi, MDB_val *key, MDB_val *data) { + MDB_cursor mc; + MDB_xcursor mx; + MDB_cursor_op op; + MDB_val rdata, *xdata; int rc, exact; - unsigned int ki; - MDB_node *leaf; - MDB_pageparent mpp; DKBUF; assert(key != NULL); @@ -3511,67 +3916,29 @@ mdb_del(MDB_txn *txn, MDB_dbi dbi, return EINVAL; } - mpp.mp_parent = NULL; - mpp.mp_pi = 0; - if ((rc = mdb_search_page(txn, dbi, key, NULL, 1, &mpp)) != MDB_SUCCESS) - return rc; - - leaf = mdb_search_node(txn, dbi, mpp.mp_page, key, &exact, &ki); - if (leaf == NULL || !exact) { - return MDB_NOTFOUND; + mc.mc_txn = txn; + mc.mc_dbi = dbi; + mc.mc_flags = 0; + if (txn->mt_dbs[dbi].md_flags & MDB_DUPSORT) { + mc.mc_xcursor = &mx; + mdb_xcursor_init0(&mc); + } else { + mc.mc_xcursor = NULL; } - if (!IS_LEAF2(mpp.mp_page) && F_ISSET(leaf->mn_flags, F_DUPDATA)) { - MDB_xcursor mx; - MDB_pageparent mp2; - - mdb_xcursor_init0(txn, dbi, &mx); - mdb_xcursor_init1(txn, dbi, &mx, mpp.mp_page, leaf); - if (data) { - rc = mdb_del(&mx.mx_txn, mx.mx_cursor.mc_dbi, data, NULL); - mdb_xcursor_fini(txn, dbi, &mx); - /* If sub-DB still has entries, we're done */ - if (mx.mx_txn.mt_dbs[mx.mx_cursor.mc_dbi].md_root != P_INVALID) { - memcpy(NODEDATA(leaf), &mx.mx_txn.mt_dbs[mx.mx_cursor.mc_dbi], - sizeof(MDB_db)); - txn->mt_dbs[dbi].md_entries--; - return rc; - } - /* otherwise fall thru and delete the sub-DB */ - } else { - /* add all the child DB's pages to the free list */ - rc = mdb_search_page(&mx.mx_txn, mx.mx_cursor.mc_dbi, - NULL, &mx.mx_cursor, 0, &mp2); - if (rc == MDB_SUCCESS) { - MDB_ppage *top, *parent; - MDB_node *ni; - unsigned int i; - - cursor_pop_page(&mx.mx_cursor); - if (mx.mx_cursor.mc_snum) { - top = CURSOR_TOP(&mx.mx_cursor); - while (mx.mx_cursor.mc_snum > 1) { - parent = CURSOR_PARENT(&mx.mx_cursor); - for (i=0; imp_page); i++) { - ni = NODEPTR(top->mp_page, i); - mdb_midl_insert(txn->mt_free_pgs, NODEPGNO(ni)); - } - parent->mp_ki++; - if (parent->mp_ki >= NUMKEYS(parent->mp_page)) { - cursor_pop_page(&mx.mx_cursor); - top = parent; - } else { - ni = NODEPTR(parent->mp_page, parent->mp_ki); - rc = mdb_get_page(&mx.mx_txn, NODEPGNO(ni), &top->mp_page); - } - } - } - mdb_midl_insert(txn->mt_free_pgs, mx.mx_txn.mt_dbs[mx.mx_cursor.mc_dbi].md_root); - } - } + exact = 0; + if (data) { + op = MDB_GET_BOTH; + rdata = *data; + xdata = &rdata; + } else { + op = MDB_SET; + xdata = NULL; } - - return mdb_del0(txn, dbi, ki, &mpp, leaf); + rc = mdb_cursor_set(&mc, key, xdata, op, &exact); + if (rc == 0) + rc = mdb_cursor_del(&mc, data ? 0 : MDB_NODUPDATA); + return rc; } /* Split page <*mpp>, and insert in either left or @@ -3716,7 +4083,6 @@ mdb_split(MDB_txn *txn, MDB_dbi dbi, MDB_page **mpp, unsigned int *newindxp, /* First find the separating key between the split pages. */ - memset(&sepkey, 0, sizeof(sepkey)); if (newindx == split_indx) { sepkey.mv_size = newkey->mv_size; sepkey.mv_data = newkey->mv_data; @@ -3823,6 +4189,7 @@ newsep: return rc; } +#if 0 static int mdb_put0(MDB_txn *txn, MDB_dbi dbi, MDB_val *key, MDB_val *data, unsigned int flags) @@ -3946,11 +4313,16 @@ new_sub: * DB are all zero size. */ if (do_sub) { + MDB_cursor mc; MDB_xcursor mx; leaf = NODEPTR(mpp.mp_page, ki); put_sub: - mdb_xcursor_init0(txn, dbi, &mx); + mc.mc_txn = txn; + mc.mc_dbi = dbi; + mc.mc_flags = 0; + mc.mc_xcursor = &mx; + mdb_xcursor_init0(&mc); mdb_xcursor_init1(txn, dbi, &mx, mpp.mp_page, leaf); xdata.mv_size = 0; xdata.mv_data = ""; @@ -3963,7 +4335,7 @@ put_sub: leaf->mn_flags |= F_DUPDATA; } rc = mdb_put0(&mx.mx_txn, mx.mx_cursor.mc_dbi, data, &xdata, flags); - mdb_xcursor_fini(txn, dbi, &mx); + mdb_xcursor_fini(&mc); memcpy(NODEDATA(leaf), &mx.mx_txn.mt_dbs[mx.mx_cursor.mc_dbi], sizeof(MDB_db)); } @@ -3973,11 +4345,16 @@ put_sub: done: return rc; } +#endif int mdb_put(MDB_txn *txn, MDB_dbi dbi, MDB_val *key, MDB_val *data, unsigned int flags) { + MDB_cursor mc; + MDB_xcursor mx; + int rc; + assert(key != NULL); assert(data != NULL); @@ -3995,7 +4372,17 @@ mdb_put(MDB_txn *txn, MDB_dbi dbi, if ((flags & (MDB_NOOVERWRITE|MDB_NODUPDATA)) != flags) return EINVAL; - return mdb_put0(txn, dbi, key, data, flags); + mc.mc_txn = txn; + mc.mc_dbi = dbi; + mc.mc_snum = 0; + mc.mc_flags = 0; + if (txn->mt_dbs[dbi].md_flags & MDB_DUPSORT) { + mc.mc_xcursor = &mx; + mdb_xcursor_init0(&mc); + } else { + mc.mc_xcursor = NULL; + } + return mdb_cursor_put(&mc, key, data, flags); } int @@ -4091,13 +4478,17 @@ int mdb_open(MDB_txn *txn, const char *name, unsigned int flags, MDB_dbi *dbi) /* Create if requested */ if (rc == MDB_NOTFOUND && (flags & MDB_CREATE)) { + MDB_cursor mc; MDB_db dummy; data.mv_size = sizeof(MDB_db); data.mv_data = &dummy; memset(&dummy, 0, sizeof(dummy)); dummy.md_root = P_INVALID; dummy.md_flags = flags & 0xffff; - rc = mdb_put0(txn, MAIN_DBI, &key, &data, F_SUBDATA); + mc.mc_txn = txn; + mc.mc_dbi = MAIN_DBI; + mc.mc_flags = 0; + rc = mdb_cursor_put(&mc, &key, &data, F_SUBDATA); dirty = 1; } diff --git a/libraries/libmdb/mdb.h b/libraries/libmdb/mdb.h index 0f564c5d82..e35a6a2c2f 100644 --- a/libraries/libmdb/mdb.h +++ b/libraries/libmdb/mdb.h @@ -133,7 +133,7 @@ typedef int (MDB_cmp_func)(const MDB_val *a, const MDB_val *b); */ typedef void (MDB_rel_func)(void *newptr, void *oldptr, size_t size); -/** @defgroup mdb_env environment flags +/** @defgroup mdb_env Environment Flags * @{ */ /** mmap at a fixed address */ @@ -144,7 +144,7 @@ typedef void (MDB_rel_func)(void *newptr, void *oldptr, size_t size); #define MDB_RDONLY 0x20000 /** @} */ -/** @defgroup mdb_open database flags +/** @defgroup mdb_open Database Flags * @{ */ /** use reverse string keys */ @@ -161,15 +161,18 @@ typedef void (MDB_rel_func)(void *newptr, void *oldptr, size_t size); #define MDB_CREATE 0x40000 /** @} */ -/** @defgroup mdb_put mdb_put flags +/** @defgroup mdb_put Write Flags * @{ */ -/** For mdb_put: don't write if the key already exists. */ +/** For put: Don't write if the key already exists. */ #define MDB_NOOVERWRITE 0x10 -/** For mdb_put: don't write if the key and data pair already exist. - * Only for #MDB_DUPSORT +/** Only for #MDB_DUPSORT
+ * For put: don't write if the key and data pair already exist.
+ * For mdb_cursor_del: remove all duplicate data items. */ #define MDB_NODUPDATA 0x20 +/** For mdb_cursor_put: overwrite the current key/data pair */ +#define MDB_CURRENT 0x40 /* @} */ /** Cursor operations */ @@ -723,8 +726,6 @@ int mdb_del(MDB_txn *txn, MDB_dbi dbi, MDB_val *key, MDB_val *data); /** Create a cursor handle. * Cursors are associated with a specific transaction and database and * may not span threads. - * @todo Cursors only support read operations. Support for cursor_put() and - * cursor_del() needs to be added. * @param[in] txn A transaction handle returned by #mdb_txn_begin() * @param[in] dbi A database handle returned by #mdb_open() * @param[out] cursor Address where the new #MDB_cursor handle will be stored diff --git a/libraries/libmdb/midl.h b/libraries/libmdb/midl.h index a08046308c..1e6c3c3843 100644 --- a/libraries/libmdb/midl.h +++ b/libraries/libmdb/midl.h @@ -29,12 +29,16 @@ /** @defgroup internal MDB Internals * @{ */ + /** ULONG should be the largest integer type supported on a machine. + * It should be equal to the size of a pointer. + */ +#define ULONG unsigned long /** @defgroup idls ID List Management * @{ */ - /** An ID should be the largest integer type supported on a machine. + /** A generic ID number. These were entryIDs in back-bdb. */ -#define ID unsigned long +typedef ULONG ID; /** An IDL is an ID List, a sorted array of IDs. The first * element of the array is a counter for how many actual -- 2.39.5