Sorted dup fixes

libraries/libmdb/Makefile

@@ -4,7 +4,7 @@ OPT = -O2 -g
 CFLAGS	= -pthread $(OPT) $(W) $(XCFLAGS)
 LDLIBS	=
 
-all:	mdb_stat mtest mtest2
+all:	mdb_stat mtest mtest2 mtest3
 
 clean:
 	rm -rf mtest mdb_stat *.[ao] *~ testdb
@@ -16,6 +16,7 @@ test:	all
 mdb_stat: mdb_stat.o mdb.o idl.o
 mtest:    mtest.o    mdb.o idl.o
 mtest2:	mtest2.o mdb.o idl.o
+mtest3:	mtest3.o mdb.o idl.o
 
 %:	%.o mdb.o
 	$(CC) $(CFLAGS) $(LDFLAGS) $^ $(LDLIBS) -o $@

libraries/libmdb/mdb.c

@@ -386,7 +386,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,
@@ -1678,7 +1678,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;
@@ -1801,7 +1801,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;
@@ -1823,7 +1823,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;
 		}
@@ -1859,7 +1859,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;
@@ -1916,7 +1916,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 {
@@ -1970,7 +1970,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;
@@ -2011,7 +2011,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;
@@ -2292,21 +2292,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;
@@ -2718,7 +2729,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);
@@ -3002,11 +3013,11 @@ mdb_put0(MDB_txn *txn, MDB_dbi dbi,
 		rdata = data;
 	}
 
-	if (SIZELEFT(mpp.mp_page) < mdb_leaf_size(txn->mt_env, key, data)) {
+	if (SIZELEFT(mpp.mp_page) < mdb_leaf_size(txn->mt_env, key, rdata)) {
-		rc = mdb_split(txn, dbi, &mpp.mp_page, &ki, key, data, P_INVALID);
+		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)
@@ -3031,7 +3042,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)

libraries/libmdb/mtest2.c

@@ -1,4 +1,4 @@
-/* mtest.c - memory-mapped database tester/toy */
+/* mtest2.c - memory-mapped database tester/toy */
 /*
  * Copyright 2011 Howard Chu, Symas Corp.
  * All rights reserved.
@@ -11,6 +11,9 @@
  * top-level directory of the distribution or, alternatively, at
  * <http://www.OpenLDAP.org/license.html>.
  */
+
+/* Just like mtest.c, but using a subDB instead of the main DB */
+
 #define _XOPEN_SOURCE 500		/* srandom(), random() */
 #include <stdio.h>
 #include <stdlib.h>

libraries/libmdb/mtest3.c

@@ -0,0 +1,127 @@
+/* mtest3.c - memory-mapped database tester/toy */
+/*
+ * Copyright 2011 Howard Chu, Symas Corp.
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted only as authorized by the OpenLDAP
+ * Public License.
+ *
+ * A copy of this license is available in the file LICENSE in the
+ * top-level directory of the distribution or, alternatively, at
+ * <http://www.OpenLDAP.org/license.html>.
+ */
+
+/* Tests for sorted duplicate DBs */
+#define _XOPEN_SOURCE 500		/* srandom(), random() */
+#include <stdio.h>
+#include <stdlib.h>
+#include <string.h>
+#include <time.h>
+#include "mdb.h"
+
+int main(int argc,char * argv[])
+{
+	int i = 0, j = 0, rc;
+	MDB_env *env;
+	MDB_dbi dbi;
+	MDB_val key, data;
+	MDB_txn *txn;
+	MDB_stat mst;
+	MDB_cursor *cursor;
+	int count;
+	int *values;
+	char sval[32];
+	char kval[sizeof(int)];
+
+	srandom(time(NULL));
+
+	memset(sval, 0, sizeof(sval));
+
+	count = (random()%384) + 64;
+	values = (int *)malloc(count*sizeof(int));
+
+	for(i = 0;i<count;i++) {
+		values[i] = random()%1024;
+	}
+
+	rc = mdbenv_create(&env);
+	rc = mdbenv_set_mapsize(env, 10485760);
+	rc = mdbenv_set_maxdbs(env, 4);
+	rc = mdbenv_open(env, "./testdb", MDB_FIXEDMAP|MDB_NOSYNC, 0664);
+	rc = mdb_txn_begin(env, 0, &txn);
+	rc = mdb_open(txn, "id2", MDB_CREATE|MDB_DUPSORT, &dbi);
+
+	key.mv_size = sizeof(int);
+	key.mv_data = kval;
+	data.mv_size = sizeof(sval);
+	data.mv_data = sval;
+
+	printf("Adding %d values\n", count);
+	for (i=0;i<count;i++) {
+		if (!(i & 0x0f))
+			sprintf(kval, "%03x", values[i]);
+		sprintf(sval, "%03x %d foo bar", values[i], values[i]);
+		rc = mdb_put(txn, dbi, &key, &data, MDB_NODUPDATA);
+		if (rc) j++;
+	}
+	if (j) printf("%d duplicates skipped\n", j);
+	rc = mdb_txn_commit(txn);
+	rc = mdbenv_stat(env, &mst);
+
+	rc = mdb_txn_begin(env, 1, &txn);
+	rc = mdb_cursor_open(txn, dbi, &cursor);
+	while ((rc = mdb_cursor_get(cursor, &key, &data, MDB_NEXT)) == 0) {
+		printf("key: %p %.*s, data: %p %.*s\n",
+			key.mv_data,  (int) key.mv_size,  (char *) key.mv_data,
+			data.mv_data, (int) data.mv_size, (char *) data.mv_data);
+	}
+	mdb_cursor_close(cursor);
+	mdb_txn_abort(txn);
+
+	j=0;
+
+	for (i= count - 1; i > -1; i-= (random()%5)) {
+		j++;
+		txn=NULL;
+		rc = mdb_txn_begin(env, 0, &txn);
+		sprintf(kval, "%03x", values[i & ~0x0f]);
+		sprintf(sval, "%03x %d foo bar", values[i], values[i]);
+		key.mv_size = sizeof(int);
+		key.mv_data = kval;
+		data.mv_size = sizeof(sval);
+		data.mv_data = sval;
+		rc = mdb_del(txn, dbi, &key, &data, MDB_DEL_DUP);
+		if (rc) {
+			j--;
+			mdb_txn_abort(txn);
+		} else {
+			rc = mdb_txn_commit(txn);
+		}
+	}
+	free(values);
+	printf("Deleted %d values\n", j);
+
+	rc = mdbenv_stat(env, &mst);
+	rc = mdb_txn_begin(env, 1, &txn);
+	rc = mdb_cursor_open(txn, dbi, &cursor);
+	printf("Cursor next\n");
+	while ((rc = mdb_cursor_get(cursor, &key, &data, MDB_NEXT)) == 0) {
+		printf("key: %.*s, data: %.*s\n",
+			(int) key.mv_size,  (char *) key.mv_data,
+			(int) data.mv_size, (char *) data.mv_data);
+	}
+	printf("Cursor prev\n");
+	while ((rc = mdb_cursor_get(cursor, &key, &data, MDB_PREV)) == 0) {
+		printf("key: %.*s, data: %.*s\n",
+			(int) key.mv_size,  (char *) key.mv_data,
+			(int) data.mv_size, (char *) data.mv_data);
+	}
+	mdb_cursor_close(cursor);
+	mdb_close(txn, dbi);
+
+	mdb_txn_abort(txn);
+	mdbenv_close(env);
+
+	return 0;
+}