More search optimization

Tighten up entry_alloc/entry_decode
Track parent nodes in idscopes

servers/slapd/back-mdb/back-mdb.h

@@ -165,17 +165,6 @@ typedef struct IndexRec {
 
 #define MAXRDNS	SLAP_LDAPDN_MAXLEN/4
 
-typedef struct IdScopes {
-	MDB_txn *mt;
-	MDB_cursor *mc;
-	ID id;
-	ID *scopes;
-	int numrdns;
-	int nscope;
-	struct berval rdns[MAXRDNS];
-	struct berval nrdns[MAXRDNS];
-} IdScopes;
-
 #include "proto-mdb.h"
 
 #endif /* _BACK_MDB_H_ */

servers/slapd/back-mdb/dn2id.c

@@ -725,6 +725,7 @@ mdb_idscopes(
 	MDB_dbi dbi = mdb->mi_dn2id;
 	MDB_val		key, data;
 	ID id;
+	ID2 id2;
 	char	*ptr;
 	int		rc;
 	unsigned int x;
@@ -740,12 +741,22 @@ mdb_idscopes(
 
 	id = isc->id;
 	while (id) {
-		key.mv_data = &id;
-		rc = mdb_cursor_get( isc->mc, &key, &data, MDB_SET );
-		if ( rc )
-			break;
+		x = mdb_id2l_search( isc->scopes, id );
+		if ( x <= isc->scopes[0].mid && isc->scopes[x].mid == id ) {
+			if ( !isc->scopes[x].mval.mv_data ) {
+				isc->nscope = x;
+				return MDB_SUCCESS;
+			}
+			data = isc->scopes[x].mval;
+			rc = 1;
+		} else {
+			key.mv_data = &id;
+			rc = mdb_cursor_get( isc->mc, &key, &data, MDB_SET );
+			if ( rc )
+				break;
 
-		/* save RDN info */
+			/* save RDN info */
+		}
 		d = data.mv_data;
 		nrlen = (d->nrdnlen[0] << 8) | d->nrdnlen[1];
 		rlen = data.mv_size - sizeof(diskNode) - nrlen;
@@ -755,14 +766,15 @@ mdb_idscopes(
 		isc->rdns[isc->numrdns].bv_val = d->nrdn+nrlen+1;
 		isc->numrdns++;
 
+
+		if (!rc && id != isc->id) {
+			id2.mid = id;
+			id2.mval = data;
+			mdb_id2l_insert( isc->scopes, &id2 );
+		}
 		ptr = data.mv_data;
 		ptr += data.mv_size - sizeof(ID);
 		memcpy( &id, ptr, sizeof(ID) );
-		x = mdb_idl_search( isc->scopes, id );
-		if ( isc->scopes[x] == id ) {
-			isc->nscope = x;
-			return MDB_SUCCESS;
-		}
 		if ( op->ors_scope == LDAP_SCOPE_ONELEVEL )
 			break;
 	}

servers/slapd/back-mdb/id2entry.c

@@ -122,6 +122,7 @@ int mdb_id2entry(
 			*bptr++ = gluebv;
 			BER_BVZERO(bptr);
 			bptr++;
+			a->a_next = a+1;
 			a = a->a_next;
 			a->a_flags = SLAP_ATTR_DONT_FREE_DATA | SLAP_ATTR_DONT_FREE_VALS;
 			a->a_desc = slap_schema.si_ad_structuralObjectClass;
@@ -130,6 +131,7 @@ int mdb_id2entry(
 			a->a_numvals = 1;
 			*bptr++ = gluebv;
 			BER_BVZERO(bptr);
+			a->a_next = NULL;
 			*e = r;
 			return MDB_SUCCESS;
 		}
@@ -170,19 +172,13 @@ static Entry * mdb_entry_alloc(
 	int nattrs,
 	int nvals )
 {
-	Attribute *a;
 	Entry *e = op->o_tmpalloc( sizeof(Entry) +
 		nattrs * sizeof(Attribute) +
 		nvals * sizeof(struct berval), op->o_tmpmemctx );
 	BER_BVZERO(&e->e_bv);
 	e->e_private = e;
 	e->e_attrs = (Attribute *)(e+1);
-	for (a=e->e_attrs; nattrs>1; nattrs--) {
-		a->a_next = a+1;
-		a++;
-	}
-	a->a_next = NULL;
-	e->e_attrs->a_vals = (struct berval *)(a+1);
+	e->e_attrs->a_vals = (struct berval *)(e->e_attrs+nattrs);
 
 	return e;
 }
@@ -674,8 +670,10 @@ int mdb_entry_decode(Operation *op, MDB_val *data, Entry **e)
 				return rc;
 			}
 		}
+		a->a_next = a+1;
 		a = a->a_next;
 	}
+	a[-1].a_next = NULL;
 done:
 
 	Debug(LDAP_DEBUG_TRACE, "<= mdb_entry_decode\n",

servers/slapd/back-mdb/idl.c

@@ -25,7 +25,7 @@
 #define IDL_MAX(x,y)	( x > y ? x : y )
 #define IDL_MIN(x,y)	( x < y ? x : y )
 
-#define IDL_CMP(x,y)	( x < y ? -1 : ( x > y ? 1 : 0 ) )
+#define IDL_CMP(x,y)	( x < y ? -1 : ( x > y ) )
 
 #if IDL_DEBUG > 0
 static void idl_check( ID *ids )
@@ -1143,3 +1143,71 @@ mdb_idl_sort( ID *ids, ID *tmp )
 	}
 }
 #endif	/* Quick vs Radix */
+
+unsigned mdb_id2l_search( ID2L ids, ID 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 = IDL_CMP( id, 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_id2l_insert( ID2L ids, ID2 *id )
+{
+	unsigned x, i;
+
+	x = mdb_id2l_search( ids, id->mid );
+	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_UM_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;
+}

servers/slapd/back-mdb/idl.h

@@ -68,7 +68,46 @@
 #define MDB_IDL_N( ids )		( MDB_IDL_IS_RANGE(ids) \
 	? (ids[2]-ids[1])+1 : ids[0] )
 
+	/** An ID2 is an ID/value pair.
+	 */
+typedef struct ID2 {
+	ID mid;			/**< The ID */
+	MDB_val mval;		/**< The value */
+} ID2;
+
+	/** An ID2L is an ID2 List, a sorted array of ID2s.
+	 * The first element's \b mid member is a count of how many actual
+	 * elements are in the array. The \b mptr member of the first element is unused.
+	 * The array is sorted in ascending order by \b mid.
+	 */
+typedef ID2 *ID2L;
+
+typedef struct IdScopes {
+	MDB_txn *mt;
+	MDB_cursor *mc;
+	ID id;
+	ID2L scopes;
+	int numrdns;
+	int nscope;
+	struct berval rdns[MAXRDNS];
+	struct berval nrdns[MAXRDNS];
+} IdScopes;
+
 LDAP_BEGIN_DECL
+	/** Search for an ID in an ID2L.
+	 * @param[in] ids	The ID2L to search.
+	 * @param[in] id	The ID to search for.
+	 * @return	The index of the first ID2 whose \b mid member is greater than or equal to \b id.
+	 */
+unsigned mdb_id2l_search( ID2L ids, ID id );
+
+
+	/** Insert an ID2 into a ID2L.
+	 * @param[in,out] ids	The ID2L to insert into.
+	 * @param[in] id	The ID2 to insert.
+	 * @return	0 on success, -1 if the ID was already present in the MIDL2.
+	 */
+int mdb_id2l_insert( ID2L ids, ID2 *id );
 LDAP_END_DECL
 
 #endif

servers/slapd/back-mdb/proto-mdb.h

@@ -127,9 +127,11 @@ int mdb_idscope(
 	ID *ids,
 	ID *res );
 
+struct IdScopes;
+
 int mdb_idscopes(
 	Operation *op,
-	IdScopes *isc );
+	struct IdScopes *isc );
 
 MDB_cmp_func mdb_dup_compare;
 

servers/slapd/back-mdb/search.c

@@ -34,7 +34,7 @@ static int search_candidates(
 	MDB_txn *txn,
 	MDB_cursor *mci,
 	ID	*ids,
-	ID	*scopes );
+	ID2L	scopes );
 
 static int parse_paged_cookie( Operation *op, SlapReply *rs );
 
@@ -132,7 +132,7 @@ static int search_aliases(
 	Entry *e,
 	MDB_txn *txn,
 	MDB_cursor *mci,
-	ID *scopes,
+	ID2L scopes,
 	ID *stack )
 {
 	ID *aliases, *curscop, *visited, *newsubs, *oldsubs, *tmp;
@@ -210,7 +210,10 @@ static int search_aliases(
 				/* If the target was not already in our current scopes,
 				 * make note of it in the newsubs list.
 				 */
-				if (mdb_idl_insert(scopes, a->e_id) == 0) {
+				ID2 mid;
+				mid.mid = a->e_id;
+				mid.mval.mv_data = NULL;
+				if (mdb_id2l_insert(scopes, &mid) == 0) {
 					mdb_idl_insert(newsubs, a->e_id);
 				}
 				mdb_entry_return( op, a );
@@ -280,7 +283,7 @@ mdb_search( Operation *op, SlapReply *rs )
 	ID		id, cursor;
 	ID		lastid = NOID;
 	ID		candidates[MDB_IDL_UM_SIZE];
-	ID		scopes[MDB_IDL_DB_SIZE];
+	ID2		scopes[MDB_IDL_UM_SIZE];
 	Entry		*e = NULL, *base = NULL;
 	Entry		*matched = NULL;
 	AttributeName	*attrs;
@@ -490,8 +493,9 @@ dn2entry_retry:
 
 	} else {
 		MDB_IDL_ZERO( candidates );
-		MDB_IDL_ZERO( scopes );
-		mdb_idl_insert( scopes, base->e_id );
+		scopes[0].mid = 1;
+		scopes[1].mid = base->e_id;
+		scopes[1].mval.mv_data = NULL;
 		rs->sr_err = search_candidates( op, rs, base,
 			ltid, mci, candidates, scopes );
 	}
@@ -717,7 +721,7 @@ loop_begin:
 				pdn = base->e_name;
 				pndn = base->e_nname;
 			} else {
-				mdb_id2name( op, ltid, &isc.mc, scopes[isc.nscope], &pdn, &pndn );
+				mdb_id2name( op, ltid, &isc.mc, scopes[isc.nscope].mid, &pdn, &pndn );
 			}
 			e->e_name.bv_len = pdn.bv_len;
 			e->e_nname.bv_len = pndn.bv_len;
@@ -966,7 +970,7 @@ static int search_candidates(
 	MDB_txn *txn,
 	MDB_cursor *mci,
 	ID	*ids,
-	ID	*scopes )
+	ID2L	scopes )
 {
 	struct mdb_info *mdb = (struct mdb_info *) op->o_bd->be_private;
 	int rc, depth = 1;