From: Howard Chu <hyc@openldap.org>
Date: Sun, 24 Apr 2005 22:55:21 +0000 (+0000)
Subject: ITS#3671 free all locks before calling send_search_*, queue entries if
X-Git-Tag: OPENLDAP_AC_BP~806
X-Git-Url: https://git.sur5r.net/?a=commitdiff_plain;h=895b74cb1dd5f67a5b56b71bebf6f92bf4d0e478;p=openldap

ITS#3671 free all locks before calling send_search_*, queue entries if
the connection is busy. Probably need to set a queue limit here and
abandon the psearch if the limit is reached.
---

diff --git a/servers/slapd/overlays/syncprov.c b/servers/slapd/overlays/syncprov.c
index 7ff8df6dd0..5bf46f98b7 100644
--- a/servers/slapd/overlays/syncprov.c
+++ b/servers/slapd/overlays/syncprov.c
@@ -131,6 +131,7 @@ typedef struct opcookie {
 	struct berval sndn;
 	struct berval suuid;	/* UUID of entry */
 	struct berval sctxcsn;
+	syncops	*shead;		/* head of si_ops when we started */
 	int sreference;	/* Is the entry a reference? */
 } opcookie;
 
@@ -660,7 +661,7 @@ syncprov_findcsn( Operation *op, int mode )
 	return rc;
 }
 
-/* Queue a persistent search response if still in Refresh stage */
+/* Queue a persistent search response */
 static int
 syncprov_qresp( opcookie *opc, syncops *so, int mode )
 {
@@ -670,13 +671,18 @@ syncprov_qresp( opcookie *opc, syncops *so, int mode )
 		opc->sdn.bv_len + 1 + opc->sndn.bv_len + 1 + opc->sctxcsn.bv_len + 1 );
 	sr->s_next = NULL;
 	sr->s_dn.bv_val = (char *)(sr + 1);
+	sr->s_dn.bv_len = opc->sdn.bv_len;
 	sr->s_mode = mode;
 	sr->s_isreference = opc->sreference;
 	sr->s_ndn.bv_val = lutil_strcopy( sr->s_dn.bv_val, opc->sdn.bv_val );
+	sr->s_ndn.bv_len = opc->sndn.bv_len;
 	*(sr->s_ndn.bv_val++) = '\0';
 	sr->s_uuid.bv_val = lutil_strcopy( sr->s_ndn.bv_val, opc->sndn.bv_val );
+	sr->s_uuid.bv_len = opc->suuid.bv_len;
 	*(sr->s_uuid.bv_val++) = '\0';
 	sr->s_csn.bv_val = lutil_strcopy( sr->s_uuid.bv_val, opc->suuid.bv_val );
+	sr->s_csn.bv_len = opc->sctxcsn.bv_len;
+	strcpy( sr->s_csn.bv_val, opc->sctxcsn.bv_val );
 
 	if ( !so->s_res ) {
 		so->s_res = sr;
@@ -688,9 +694,58 @@ syncprov_qresp( opcookie *opc, syncops *so, int mode )
 	return LDAP_SUCCESS;
 }
 
+/* Play back queued responses */
+static int
+syncprov_sendresp( Operation *op, opcookie *opc, syncops *so, Entry **e, int mode, int queue );
+
+static int
+syncprov_qplay( Operation *op, slap_overinst *on, syncops *so )
+{
+	syncres *sr, *srnext;
+	Entry *e;
+	opcookie opc;
+	int rc;
+
+	opc.son = on;
+	op->o_bd->bd_info = (BackendInfo *)on->on_info;
+	for (sr = so->s_res; sr; sr=srnext) {
+		int rc;
+		srnext = sr->s_next;
+		opc.sdn = sr->s_dn;
+		opc.sndn = sr->s_ndn;
+		opc.suuid = sr->s_uuid;
+		opc.sctxcsn = sr->s_csn;
+		opc.sreference = sr->s_isreference;
+		e = NULL;
+
+		if ( sr->s_mode != LDAP_SYNC_DELETE ) {
+			rc = be_entry_get_rw( op, &opc.sndn, NULL, NULL, 0, &e );
+			if ( rc ) {
+				ch_free( sr );
+				so->s_res = srnext;
+				continue;
+			}
+		}
+		rc = syncprov_sendresp( op, &opc, so, &e, sr->s_mode, 0 );
+
+		if ( e ) {
+			be_entry_release_rw( op, e, 0 );
+		}
+		if ( rc )
+			break;
+
+		ch_free( sr );
+		so->s_res = srnext;
+	}
+	op->o_bd->bd_info = (BackendInfo *)on;
+	if ( !so->s_res )
+		so->s_restail = NULL;
+	return rc;
+}
+
 /* Send a persistent search response */
 static int
-syncprov_sendresp( Operation *op, opcookie *opc, syncops *so, Entry *e, int mode, int queue )
+syncprov_sendresp( Operation *op, opcookie *opc, syncops *so, Entry **e, int mode, int queue )
 {
 	slap_overinst *on = opc->son;
 	syncprov_info_t *si = on->on_bi.bi_private;
@@ -708,12 +763,30 @@ syncprov_sendresp( Operation *op, opcookie *opc, syncops *so, Entry *e, int mode
 	sop.o_tmpmemctx = op->o_tmpmemctx;
 	sop.o_bd = op->o_bd;
 	sop.o_controls = op->o_controls;
+	sop.o_private = op->o_private;
 
-	if ( queue && (so->s_flags & PS_IS_REFRESHING) ) {
+	/* If queueing is allowed */
+	if ( queue ) {
 		ldap_pvt_thread_mutex_lock( &so->s_mutex );
-		if ( so->s_flags & PS_IS_REFRESHING )
+		/* If we're still in refresh mode, must queue */
+		if (so->s_flags & PS_IS_REFRESHING) {
 			return syncprov_qresp( opc, so, mode );
+		}
+		/* If connection is free but queue is non-empty,
+		 * try to flush the queue.
+		 */
+		if ( so->s_res ) {
+			rs.sr_err = syncprov_qplay( &sop, on, so );
+		}
+		/* If the connection is busy, must queue */
+		if ( sop.o_conn->c_writewaiter || rs.sr_err == LDAP_BUSY ) {
+			return syncprov_qresp( opc, so, mode );
+		}
 		ldap_pvt_thread_mutex_unlock( &so->s_mutex );
+	} else {
+		/* Queueing not allowed and conn is busy, give up */
+		if ( sop.o_conn->c_writewaiter )
+			return LDAP_BUSY;
 	}
 
 	ctrls[1] = NULL;
@@ -725,20 +798,30 @@ syncprov_sendresp( Operation *op, opcookie *opc, syncops *so, Entry *e, int mode
 	rs.sr_err = syncprov_state_ctrl( &sop, &rs, &e_uuid,
 		mode, ctrls, 0, 1, &cookie );
 
-	rs.sr_entry = e;
 	rs.sr_ctrls = ctrls;
+	op->o_bd->bd_info = (BackendInfo *)on->on_info;
 	switch( mode ) {
 	case LDAP_SYNC_ADD:
+		rs.sr_entry = *e;
+		if ( rs.sr_entry->e_private )
+			rs.sr_flags = REP_ENTRY_MUSTRELEASE;
 		if ( opc->sreference ) {
-			rs.sr_ref = get_entry_referrals( &sop, e );
+			rs.sr_ref = get_entry_referrals( &sop, rs.sr_entry );
 			send_search_reference( &sop, &rs );
 			ber_bvarray_free( rs.sr_ref );
+			if ( !rs.sr_entry )
+				*e = NULL;
 			break;
 		}
 		/* fallthru */
 	case LDAP_SYNC_MODIFY:
+		rs.sr_entry = *e;
+		if ( rs.sr_entry->e_private )
+			rs.sr_flags = REP_ENTRY_MUSTRELEASE;
 		rs.sr_attrs = sop.ors_attrs;
 		send_search_entry( &sop, &rs );
+		if ( !rs.sr_entry )
+			*e = NULL;
 		break;
 	case LDAP_SYNC_DELETE:
 		e_uuid.e_attrs = NULL;
@@ -757,6 +840,7 @@ syncprov_sendresp( Operation *op, opcookie *opc, syncops *so, Entry *e, int mode
 		assert(0);
 	}
 	op->o_tmpfree( rs.sr_ctrls[0], op->o_tmpmemctx );
+	op->o_private = sop.o_private;
 	rs.sr_ctrls = NULL;
 	return rs.sr_err;
 }
@@ -902,8 +986,11 @@ syncprov_matchops( Operation *op, opcookie *opc, int saveit )
 			ber_dupbv_x( &opc->suuid, &a->a_nvals[0], op->o_tmpmemctx );
 	}
 
-	ldap_pvt_thread_mutex_lock( &si->si_ops_mutex );
-	for (ss = si->si_ops, sprev = (syncops *)&si->si_ops; ss;
+	if (saveit) {
+		ldap_pvt_thread_mutex_lock( &si->si_ops_mutex );
+		opc->shead = si->si_ops;
+	}
+	for (ss = opc->shead, sprev = (syncops *)&si->si_ops; ss;
 		sprev = ss, ss=snext)
 	{
 		syncmatches *sm;
@@ -951,7 +1038,7 @@ syncprov_matchops( Operation *op, opcookie *opc, int saveit )
 				opc->smatches = sm;
 			} else {
 				/* if found send UPDATE else send ADD */
-				syncprov_sendresp( op, opc, ss, e,
+				syncprov_sendresp( op, opc, ss, &e,
 					found ? LDAP_SYNC_MODIFY : LDAP_SYNC_ADD, 1 );
 			}
 		} else if ( !saveit && found ) {
@@ -959,9 +1046,10 @@ syncprov_matchops( Operation *op, opcookie *opc, int saveit )
 			syncprov_sendresp( op, opc, ss, NULL, LDAP_SYNC_DELETE, 1 );
 		}
 	}
-	ldap_pvt_thread_mutex_unlock( &si->si_ops_mutex );
+	if (saveit)
+		ldap_pvt_thread_mutex_unlock( &si->si_ops_mutex );
 done:
-	if ( op->o_tag != LDAP_REQ_ADD ) {
+	if ( op->o_tag != LDAP_REQ_ADD && e ) {
 		op->o_bd->bd_info = (BackendInfo *)on->on_info;
 		be_entry_release_rw( op, e, 0 );
 		op->o_bd->bd_info = (BackendInfo *)on;
@@ -1308,14 +1396,12 @@ syncprov_op_response( Operation *op, SlapReply *rs )
 				/* for each match in opc->smatches:
 				 *   send DELETE msg
 				 */
-				ldap_pvt_thread_mutex_lock( &si->si_ops_mutex );
 				for ( sm = opc->smatches; sm; sm=sm->sm_next ) {
 					if ( sm->sm_op->s_op->o_abandon )
 						continue;
 					syncprov_sendresp( op, opc, sm->sm_op, NULL,
 						LDAP_SYNC_DELETE, 1 );
 				}
-				ldap_pvt_thread_mutex_unlock( &si->si_ops_mutex );
 				break;
 			}
 		}
@@ -1469,7 +1555,6 @@ syncprov_op_mod( Operation *op, SlapReply *rs )
 
 	if (( si->si_ops || si->si_logs ) && op->o_tag != LDAP_REQ_ADD )
 		syncprov_matchops( op, opc, 1 );
-		
 
 	return SLAP_CB_CONTINUE;
 }
@@ -1633,41 +1718,9 @@ syncprov_search_response( Operation *op, SlapReply *rs )
 				&cookie, 1, NULL, 0 );
 			/* Flush any queued persist messages */
 			if ( ss->ss_so->s_res ) {
-				syncres *sr, *srnext;
-				Entry *e;
-				opcookie opc;
-
-				opc.son = on;
 				ldap_pvt_thread_mutex_lock( &ss->ss_so->s_mutex );
 				locked = 1;
-				for (sr = ss->ss_so->s_res; sr; sr=srnext) {
-					int rc = LDAP_SUCCESS;
-					srnext = sr->s_next;
-					opc.sdn = sr->s_dn;
-					opc.sndn = sr->s_ndn;
-					opc.suuid = sr->s_uuid;
-					opc.sctxcsn = sr->s_csn;
-					opc.sreference = sr->s_isreference;
-					e = NULL;
-
-					if ( sr->s_mode != LDAP_SYNC_DELETE ) {
-						op->o_bd->bd_info = (BackendInfo *)on->on_info;
-						rc = be_entry_get_rw( op, &opc.sndn, NULL, NULL, 0, &e );
-						op->o_bd->bd_info = (BackendInfo *)on;
-					}
-					if ( rc == LDAP_SUCCESS )
-						syncprov_sendresp( op, &opc, ss->ss_so, e,
-							sr->s_mode, 0 );
-
-					if ( e ) {
-						op->o_bd->bd_info = (BackendInfo *)on->on_info;
-						be_entry_release_rw( op, e, 0 );
-						op->o_bd->bd_info = (BackendInfo *)on;
-					}
-					ch_free( sr );
-				}
-				ss->ss_so->s_res = NULL;
-				ss->ss_so->s_restail = NULL;
+				syncprov_qplay( op, on, ss->ss_so );
 			}
 
 			/* Turn off the refreshing flag */