Use IDL2 for dirty page list

libraries/libmdb/mdb.c

@@ -203,7 +203,6 @@ typedef struct MDB_meta {			/* meta (footer) page content */
 } MDB_meta;
 
 typedef struct MDB_dhead {					/* a dirty page */
-	STAILQ_ENTRY(MDB_dpage)	 md_next;	/* queue of dirty pages */
 	MDB_page	*md_parent;
 	unsigned	md_pi;				/* parent index */
 	int			md_num;
@@ -214,8 +213,6 @@ typedef struct MDB_dpage {
 	MDB_page	p;
 } MDB_dpage;
 
-STAILQ_HEAD(dirty_queue, MDB_dpage);	/* FIXME: use a sorted data structure */
-
 typedef struct MDB_oldpages {
 	struct MDB_oldpages *mo_next;
 	ULONG		mo_txnid;
@@ -289,7 +286,7 @@ struct MDB_txn {
 	MDB_env		*mt_env;	
 	pgno_t		*mt_free_pgs;	/* this is an IDL */
 	union {
-		struct dirty_queue	*dirty_queue;	/* modified pages */
+		MIDL2	*dirty_list;	/* modified pages */
 		MDB_reader	*reader;
 	} mt_u;
 	MDB_dbx		*mt_dbxs;		/* array */
@@ -333,6 +330,7 @@ struct MDB_env {
 	MDB_oldpages *me_pghead;
 	pthread_key_t	me_txkey;	/* thread-key for readers */
 	pgno_t		me_free_pgs[MDB_IDL_UM_SIZE];
+	MIDL2		me_dirty_list[MDB_IDL_DB_SIZE];
 };
 
 #define NODESIZE	 offsetof(MDB_node, mn_data)
@@ -483,6 +481,7 @@ mdb_alloc_page(MDB_txn *txn, MDB_page *parent, unsigned int parent_idx, int num)
 	MDB_dpage *dp;
 	pgno_t pgno = P_INVALID;
 	ULONG oldest;
+	MIDL2 mid;
 
 	if (txn->mt_txnid > 2) {
 
@@ -574,13 +573,15 @@ mdb_alloc_page(MDB_txn *txn, MDB_page *parent, unsigned int parent_idx, int num)
 	dp->h.md_num = num;
 	dp->h.md_parent = parent;
 	dp->h.md_pi = parent_idx;
-	STAILQ_INSERT_TAIL(txn->mt_u.dirty_queue, dp, h.md_next);
 	if (pgno == P_INVALID) {
 		dp->p.mp_pgno = txn->mt_next_pgno;
 		txn->mt_next_pgno += num;
 	} else {
 		dp->p.mp_pgno = pgno;
 	}
+	mid.mid = dp->p.mp_pgno;
+	mid.mptr = dp;
+	mdb_midl2_insert(txn->mt_u.dirty_list, &mid);
 
 	return dp;
 }
@@ -640,17 +641,13 @@ mdb_txn_begin(MDB_env *env, int rdonly, MDB_txn **ret)
 	if (rdonly) {
 		txn->mt_flags |= MDB_TXN_RDONLY;
 	} else {
-		txn->mt_u.dirty_queue = calloc(1, sizeof(*txn->mt_u.dirty_queue));
+		txn->mt_u.dirty_list = env->me_dirty_list;
-		if (txn->mt_u.dirty_queue == NULL) {
+		txn->mt_u.dirty_list[0].mid = 0;
-			free(txn);
+		txn->mt_free_pgs = env->me_free_pgs;
-			return ENOMEM;
+		txn->mt_free_pgs[0] = 0;
-		}
-		STAILQ_INIT(txn->mt_u.dirty_queue);
 
 		pthread_mutex_lock(&env->me_txns->mti_wmutex);
 		env->me_txns->mti_txnid++;
-		txn->mt_free_pgs = env->me_free_pgs;
-		txn->mt_free_pgs[0] = 0;
 	}
 
 	txn->mt_txnid = env->me_txns->mti_txnid;
@@ -712,7 +709,6 @@ mdb_txn_begin(MDB_env *env, int rdonly, MDB_txn **ret)
 void
 mdb_txn_abort(MDB_txn *txn)
 {
-	MDB_dpage *dp;
 	MDB_env	*env;
 
 	if (txn == NULL)
@@ -731,12 +727,8 @@ mdb_txn_abort(MDB_txn *txn)
 		unsigned int i;
 
 		/* Discard all dirty pages. */
-		while (!STAILQ_EMPTY(txn->mt_u.dirty_queue)) {
+		for (i=1; i<=txn->mt_u.dirty_list[0].mid; i++)
-			dp = STAILQ_FIRST(txn->mt_u.dirty_queue);
+			free(txn->mt_u.dirty_list[i].mptr);
-			STAILQ_REMOVE_HEAD(txn->mt_u.dirty_queue, h.md_next);
-			free(dp);
-		}
-		free(txn->mt_u.dirty_queue);
 
 		while ((mop = txn->mt_env->me_pghead)) {
 			txn->mt_env->me_pghead = mop->mo_next;
@@ -788,7 +780,7 @@ mdb_txn_commit(MDB_txn *txn)
 		return EINVAL;
 	}
 
-	if (STAILQ_EMPTY(txn->mt_u.dirty_queue))
+	if (!txn->mt_u.dirty_list[0].mid)
 		goto done;
 
 	DPRINTF("committing transaction %lu on mdbenv %p, root page %lu",
@@ -861,7 +853,8 @@ mdb_txn_commit(MDB_txn *txn)
 		n = 0;
 		done = 1;
 		size = 0;
-		STAILQ_FOREACH(dp, txn->mt_u.dirty_queue, h.md_next) {
+		for (i=1; i<=txn->mt_u.dirty_list[0].mid; i++) {
+			dp = txn->mt_u.dirty_list[i].mptr;
 			if (dp->p.mp_pgno != next) {
 				if (n) {
 					DPRINTF("committing %u dirty pages", n);
@@ -913,11 +906,8 @@ mdb_txn_commit(MDB_txn *txn)
 
 	/* Drop the dirty pages.
 	 */
-	while (!STAILQ_EMPTY(txn->mt_u.dirty_queue)) {
+	for (i=1; i<=txn->mt_u.dirty_list[0].mid; i++)
-		dp = STAILQ_FIRST(txn->mt_u.dirty_queue);
+		free(txn->mt_u.dirty_list[i].mptr);
-		STAILQ_REMOVE_HEAD(txn->mt_u.dirty_queue, h.md_next);
-		free(dp);
-	}
 
 	if ((n = mdb_env_sync(env, 0)) != 0 ||
 	    (n = mdb_env_write_meta(txn)) != MDB_SUCCESS) {
@@ -948,7 +938,6 @@ mdb_txn_commit(MDB_txn *txn)
 	}
 
 	pthread_mutex_unlock(&env->me_txns->mti_wmutex);
-	free(txn->mt_u.dirty_queue);
 	free(txn);
 	txn = NULL;
 
@@ -1529,14 +1518,16 @@ mdb_get_page(MDB_txn *txn, pgno_t pgno)
 	MDB_page *p = NULL;
 	int found = 0;
 
-	if (!F_ISSET(txn->mt_flags, MDB_TXN_RDONLY) && !STAILQ_EMPTY(txn->mt_u.dirty_queue)) {
+	if (!F_ISSET(txn->mt_flags, MDB_TXN_RDONLY) && txn->mt_u.dirty_list[0].mid) {
 		MDB_dpage *dp;
-		STAILQ_FOREACH(dp, txn->mt_u.dirty_queue, h.md_next) {
+		MIDL2 id;
-			if (dp->p.mp_pgno == pgno) {
+		unsigned x;
-				p = &dp->p;
+		id.mid = pgno;
-				found = 1;
+		x = mdb_midl2_search(txn->mt_u.dirty_list, &id);
-				break;
+		if (x <= txn->mt_u.dirty_list[0].mid && txn->mt_u.dirty_list[x].mid == pgno) {
-			}
+			dp = txn->mt_u.dirty_list[x].mptr;
+			p = &dp->p;
+			found = 1;
 		}
 	}
 	if (!found) {

libraries/libmdb/midl.c

@@ -24,6 +24,9 @@ typedef unsigned long pgno_t;
 /* Sort the IDLs from highest to lowest */
 #define IDL_CMP(x,y)	 ( x > y ? -1 : ( x < y ? 1 : 0 ) )
 
+/* Sort the IDL2s from lowest to highest */
+#define IDL2_CMP(x,y)	 ( x < y ? -1 : ( x > y ? 1 : 0 ) )
+
 unsigned mdb_midl_search( ID *ids, ID id )
 {
 	/*
@@ -62,7 +65,7 @@ unsigned mdb_midl_search( ID *ids, ID id )
 
 int mdb_midl_insert( ID *ids, ID id )
 {
-	unsigned x;
+	unsigned x, i;
 
 	if (MDB_IDL_IS_RANGE( ids )) {
 		/* if already in range, treat as a dup */
@@ -101,9 +104,78 @@ int mdb_midl_insert( ID *ids, ID id )
 	
 	} else {
 		/* insert id */
-		AC_MEMCPY( &ids[x+1], &ids[x], (ids[0]-x) * sizeof(ID) );
+		for (i=ids[0]; i>x; i--)
+			ids[i] = ids[i-1];
 		ids[x] = id;
 	}
 
 	return 0;
 }
+
+unsigned mdb_midl2_search( MIDL2 *ids, MIDL2 *id )
+{
+	/*
+	 * binary search of id in ids
+	 * if found, returns position of id
+	 * if not found, returns first position greater than id
+	 */
+	unsigned base = 0;
+	unsigned cursor = 0;
+	int val = 0;
+	unsigned n = ids[0].mid;
+
+	while( 0 < n ) {
+		int pivot = n >> 1;
+		cursor = base + pivot;
+		val = IDL2_CMP( id->mid, ids[cursor + 1].mid );
+
+		if( val < 0 ) {
+			n = pivot;
+
+		} else if ( val > 0 ) {
+			base = cursor + 1;
+			n -= pivot + 1;
+
+		} else {
+			return cursor + 1;
+		}
+	}
+
+	if( val > 0 ) {
+		return cursor + 2;
+	} else {
+		return cursor + 1;
+	}
+}
+
+int mdb_midl2_insert( MIDL2 *ids, MIDL2 *id )
+{
+	unsigned x, i;
+
+	x = mdb_midl2_search( ids, id );
+	assert( x > 0 );
+
+	if( x < 1 ) {
+		/* internal error */
+		return -2;
+	}
+
+	if ( x <= ids[0].mid && ids[x].mid == id->mid ) {
+		/* duplicate */
+		return -1;
+	}
+
+	if ( ids[0].mid >= MDB_IDL_DB_MAX ) {
+		/* too big */
+		return -2;
+
+	} else {
+		/* insert id */
+		ids[0].mid++;
+		for (i=ids[0].mid; i>x; i--)
+			ids[i] = ids[i-1];
+		ids[x] = *id;
+	}
+
+	return 0;
+}

libraries/libmdb/midl.h

@@ -75,4 +75,12 @@
 
 int mdb_midl_insert( ID *ids, ID id );
 
+typedef struct MIDL2 {
+	ID mid;
+	void *mptr;
+} MIDL2;
+
+unsigned mdb_midl2_search( MIDL2 *ids, MIDL2 *id );
+int mdb_midl2_insert( MIDL2 *ids, MIDL2 *id );
+
 #endif	/* _MDB_MIDL_H_ */