ITS#8226 limit size of read txns in searches

doc/man/man5/slapd-mdb.5

@@ -162,6 +162,16 @@ Specify the file protection mode that newly created database
 files should have.
 The default is 0600.
 .TP
+.BI rtxnsize \ <entries>
+Specify the maximum number of entries to process in a single read
+transaction when executing a large search. Long-lived read transactions
+prevent old database pages from being reused in write transactions, and
+so can cause significant growth of the database file when there is
+heavy write traffic. This setting causes the read transaction in
+large searches to be released and reacquired after the given number
+of entries has been read, to give writers the opportunity to
+reclaim old database pages. The default is 10000.
+.TP
 .BI searchstack \ <depth>
 Specify the depth of the stack used for search filter evaluation.
 Search filters are evaluated on a stack to accommodate nested AND / OR

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

@@ -47,6 +47,9 @@ LDAP_BEGIN_DECL
 /* Default to 10MB max */
 #define DEFAULT_MAPSIZE	(10*1048576)
 
+/* Most users will never see this */
+#define DEFAULT_RTXN_SIZE      10000
+
 #ifdef LDAP_DEVEL
 #define MDB_MONITOR_IDX
 #endif /* LDAP_DEVEL */
@@ -78,6 +81,7 @@ struct mdb_info {
 	int			mi_search_stack_depth;
 	int			mi_readers;
 
+	uint32_t	mi_rtxn_size;
 	int			mi_txn_cp;
 	uint32_t	mi_txn_cp_min;
 	uint32_t	mi_txn_cp_kbyte;

servers/slapd/back-mdb/config.c

@@ -78,6 +78,11 @@ static ConfigTable mdbcfg[] = {
 		mdb_cf_gen, "( OLcfgDbAt:0.3 NAME 'olcDbMode' "
 		"DESC 'Unix permissions of database files' "
 		"SYNTAX OMsDirectoryString SINGLE-VALUE )", NULL, NULL },
+	{ "rtxnsize", "entries", 2, 2, 0, ARG_UINT|ARG_OFFSET,
+		(void *)offsetof(struct mdb_info, mi_rtxn_size),
+		"( OLcfgDbAt:12.5 NAME 'olcDbRtxnSize' "
+		"DESC 'Number of entries to process in one read transaction' "
+		"SYNTAX OMsInteger SINGLE-VALUE )", NULL, NULL },
 	{ "searchstack", "depth", 2, 2, 0, ARG_INT|ARG_MAGIC|MDB_SSTACK,
 		mdb_cf_gen, "( OLcfgDbAt:1.9 NAME 'olcDbSearchStack' "
 		"DESC 'Depth of search stack in IDLs' "
@@ -95,7 +100,7 @@ static ConfigOCs mdbocs[] = {
 		"MUST olcDbDirectory "
 		"MAY ( olcDbCheckpoint $ olcDbEnvFlags $ "
 		"olcDbNoSync $ olcDbIndex $ olcDbMaxReaders $ olcDbMaxSize $ "
-		"olcDbMode $ olcDbSearchStack ) )",
+		"olcDbMode $ olcDbSearchStack $ olcDbRtxnSize ) )",
 		 	Cft_Database, mdbcfg },
 	{ NULL, 0, NULL }
 };

servers/slapd/back-mdb/init.c

@@ -62,6 +62,7 @@ mdb_db_init( BackendDB *be, ConfigReply *cr )
 	mdb->mi_search_stack = NULL;
 
 	mdb->mi_mapsize = DEFAULT_MAPSIZE;
+	mdb->mi_rtxn_size = DEFAULT_RTXN_SIZE;
 
 	be->be_private = mdb;
 	be->be_cf_ocs = be->bd_info->bi_cf_ocs;

servers/slapd/back-mdb/search.c

@@ -327,6 +327,7 @@ typedef struct ww_ctx {
 	ID key;
 	MDB_val data;
 	int flag;
+	int nentries;
 } ww_ctx;
 
 /* ITS#7904 if we get blocked while writing results to client,
@@ -339,21 +340,28 @@ typedef struct ww_ctx {
  * case return an LDAP_BUSY error - let the client know this search
  * couldn't succeed, but might succeed on a retry.
  */
+static void
+mdb_rtxn_snap( Operation *op, ww_ctx *ww )
+{
+	/* save cursor position and release read txn */
+	if ( ww->mcd ) {
+		MDB_val key, data;
+		mdb_cursor_get( ww->mcd, &key, &data, MDB_GET_CURRENT );
+		memcpy( &ww->key, key.mv_data, sizeof(ID) );
+		ww->data.mv_size = data.mv_size;
+		ww->data.mv_data = op->o_tmpalloc( data.mv_size, op->o_tmpmemctx );
+		memcpy(ww->data.mv_data, data.mv_data, data.mv_size);
+	}
+	mdb_txn_reset( ww->txn );
+	ww->flag = 1;
+}
+
 static void
 mdb_writewait( Operation *op, slap_callback *sc )
 {
 	ww_ctx *ww = sc->sc_private;
 	if ( !ww->flag ) {
-		if ( ww->mcd ) {
-			MDB_val key, data;
-			mdb_cursor_get( ww->mcd, &key, &data, MDB_GET_CURRENT );
-			memcpy( &ww->key, key.mv_data, sizeof(ID) );
-			ww->data.mv_size = data.mv_size;
-			ww->data.mv_data = op->o_tmpalloc( data.mv_size, op->o_tmpmemctx );
-			memcpy(ww->data.mv_data, data.mv_data, data.mv_size);
-		}
-		mdb_txn_reset( ww->txn );
-		ww->flag = 1;
+		mdb_rtxn_snap( op, ww );
 	}
 }
 
@@ -1048,14 +1056,6 @@ notfound:
 			ber_bvarray_free( erefs );
 			rs->sr_ref = NULL;
 
-			if ( wwctx.flag ) {
-				rs->sr_err = mdb_waitfixup( op, &wwctx, mci, mcd, &isc );
-				if ( rs->sr_err ) {
-					send_ldap_result( op, rs );
-					goto done;
-				}
-			}
-
 			goto loop_continue;
 		}
 
@@ -1110,13 +1110,6 @@ notfound:
 					}
 					goto done;
 				}
-				if ( wwctx.flag ) {
-					rs->sr_err = mdb_waitfixup( op, &wwctx, mci, mcd, &isc );
-					if ( rs->sr_err ) {
-						send_ldap_result( op, rs );
-						goto done;
-					}
-				}
 			}
 
 		} else {
@@ -1127,6 +1120,21 @@ notfound:
 		}
 
 loop_continue:
+		if ( moi == &opinfo && !wwctx.flag && mdb->mi_rtxn_size ) {
+			wwctx.nentries++;
+			if ( wwctx.nentries >= mdb->mi_rtxn_size ) {
+				wwctx.nentries = 0;
+				mdb_rtxn_snap( op, &wwctx );
+			}
+		}
+		if ( wwctx.flag ) {
+			rs->sr_err = mdb_waitfixup( op, &wwctx, mci, mcd, &isc );
+			if ( rs->sr_err ) {
+				send_ldap_result( op, rs );
+				goto done;
+			}
+		}
+
 		if( e != NULL ) {
 			if ( e != base )
 				mdb_entry_return( op, e );