mirror of
https://git.openldap.org/openldap/openldap.git
synced 2024-12-21 03:10:25 +08:00
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.
This commit is contained in:
parent
b92387bb1c
commit
895b74cb1d
@ -131,6 +131,7 @@ typedef struct opcookie {
|
|||||||
struct berval sndn;
|
struct berval sndn;
|
||||||
struct berval suuid; /* UUID of entry */
|
struct berval suuid; /* UUID of entry */
|
||||||
struct berval sctxcsn;
|
struct berval sctxcsn;
|
||||||
|
syncops *shead; /* head of si_ops when we started */
|
||||||
int sreference; /* Is the entry a reference? */
|
int sreference; /* Is the entry a reference? */
|
||||||
} opcookie;
|
} opcookie;
|
||||||
|
|
||||||
@ -660,7 +661,7 @@ syncprov_findcsn( Operation *op, int mode )
|
|||||||
return rc;
|
return rc;
|
||||||
}
|
}
|
||||||
|
|
||||||
/* Queue a persistent search response if still in Refresh stage */
|
/* Queue a persistent search response */
|
||||||
static int
|
static int
|
||||||
syncprov_qresp( opcookie *opc, syncops *so, int mode )
|
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 );
|
opc->sdn.bv_len + 1 + opc->sndn.bv_len + 1 + opc->sctxcsn.bv_len + 1 );
|
||||||
sr->s_next = NULL;
|
sr->s_next = NULL;
|
||||||
sr->s_dn.bv_val = (char *)(sr + 1);
|
sr->s_dn.bv_val = (char *)(sr + 1);
|
||||||
|
sr->s_dn.bv_len = opc->sdn.bv_len;
|
||||||
sr->s_mode = mode;
|
sr->s_mode = mode;
|
||||||
sr->s_isreference = opc->sreference;
|
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_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_ndn.bv_val++) = '\0';
|
||||||
sr->s_uuid.bv_val = lutil_strcopy( sr->s_ndn.bv_val, opc->sndn.bv_val );
|
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_uuid.bv_val++) = '\0';
|
||||||
sr->s_csn.bv_val = lutil_strcopy( sr->s_uuid.bv_val, opc->suuid.bv_val );
|
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 ) {
|
if ( !so->s_res ) {
|
||||||
so->s_res = sr;
|
so->s_res = sr;
|
||||||
@ -688,9 +694,58 @@ syncprov_qresp( opcookie *opc, syncops *so, int mode )
|
|||||||
return LDAP_SUCCESS;
|
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 */
|
/* Send a persistent search response */
|
||||||
static int
|
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;
|
slap_overinst *on = opc->son;
|
||||||
syncprov_info_t *si = on->on_bi.bi_private;
|
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_tmpmemctx = op->o_tmpmemctx;
|
||||||
sop.o_bd = op->o_bd;
|
sop.o_bd = op->o_bd;
|
||||||
sop.o_controls = op->o_controls;
|
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 );
|
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 );
|
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 );
|
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;
|
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,
|
rs.sr_err = syncprov_state_ctrl( &sop, &rs, &e_uuid,
|
||||||
mode, ctrls, 0, 1, &cookie );
|
mode, ctrls, 0, 1, &cookie );
|
||||||
|
|
||||||
rs.sr_entry = e;
|
|
||||||
rs.sr_ctrls = ctrls;
|
rs.sr_ctrls = ctrls;
|
||||||
|
op->o_bd->bd_info = (BackendInfo *)on->on_info;
|
||||||
switch( mode ) {
|
switch( mode ) {
|
||||||
case LDAP_SYNC_ADD:
|
case LDAP_SYNC_ADD:
|
||||||
|
rs.sr_entry = *e;
|
||||||
|
if ( rs.sr_entry->e_private )
|
||||||
|
rs.sr_flags = REP_ENTRY_MUSTRELEASE;
|
||||||
if ( opc->sreference ) {
|
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 );
|
send_search_reference( &sop, &rs );
|
||||||
ber_bvarray_free( rs.sr_ref );
|
ber_bvarray_free( rs.sr_ref );
|
||||||
|
if ( !rs.sr_entry )
|
||||||
|
*e = NULL;
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
/* fallthru */
|
/* fallthru */
|
||||||
case LDAP_SYNC_MODIFY:
|
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;
|
rs.sr_attrs = sop.ors_attrs;
|
||||||
send_search_entry( &sop, &rs );
|
send_search_entry( &sop, &rs );
|
||||||
|
if ( !rs.sr_entry )
|
||||||
|
*e = NULL;
|
||||||
break;
|
break;
|
||||||
case LDAP_SYNC_DELETE:
|
case LDAP_SYNC_DELETE:
|
||||||
e_uuid.e_attrs = NULL;
|
e_uuid.e_attrs = NULL;
|
||||||
@ -757,6 +840,7 @@ syncprov_sendresp( Operation *op, opcookie *opc, syncops *so, Entry *e, int mode
|
|||||||
assert(0);
|
assert(0);
|
||||||
}
|
}
|
||||||
op->o_tmpfree( rs.sr_ctrls[0], op->o_tmpmemctx );
|
op->o_tmpfree( rs.sr_ctrls[0], op->o_tmpmemctx );
|
||||||
|
op->o_private = sop.o_private;
|
||||||
rs.sr_ctrls = NULL;
|
rs.sr_ctrls = NULL;
|
||||||
return rs.sr_err;
|
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 );
|
ber_dupbv_x( &opc->suuid, &a->a_nvals[0], op->o_tmpmemctx );
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if (saveit) {
|
||||||
ldap_pvt_thread_mutex_lock( &si->si_ops_mutex );
|
ldap_pvt_thread_mutex_lock( &si->si_ops_mutex );
|
||||||
for (ss = si->si_ops, sprev = (syncops *)&si->si_ops; ss;
|
opc->shead = si->si_ops;
|
||||||
|
}
|
||||||
|
for (ss = opc->shead, sprev = (syncops *)&si->si_ops; ss;
|
||||||
sprev = ss, ss=snext)
|
sprev = ss, ss=snext)
|
||||||
{
|
{
|
||||||
syncmatches *sm;
|
syncmatches *sm;
|
||||||
@ -951,7 +1038,7 @@ syncprov_matchops( Operation *op, opcookie *opc, int saveit )
|
|||||||
opc->smatches = sm;
|
opc->smatches = sm;
|
||||||
} else {
|
} else {
|
||||||
/* if found send UPDATE else send ADD */
|
/* 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 );
|
found ? LDAP_SYNC_MODIFY : LDAP_SYNC_ADD, 1 );
|
||||||
}
|
}
|
||||||
} else if ( !saveit && found ) {
|
} 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 );
|
syncprov_sendresp( op, opc, ss, NULL, LDAP_SYNC_DELETE, 1 );
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
if (saveit)
|
||||||
ldap_pvt_thread_mutex_unlock( &si->si_ops_mutex );
|
ldap_pvt_thread_mutex_unlock( &si->si_ops_mutex );
|
||||||
done:
|
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;
|
op->o_bd->bd_info = (BackendInfo *)on->on_info;
|
||||||
be_entry_release_rw( op, e, 0 );
|
be_entry_release_rw( op, e, 0 );
|
||||||
op->o_bd->bd_info = (BackendInfo *)on;
|
op->o_bd->bd_info = (BackendInfo *)on;
|
||||||
@ -1308,14 +1396,12 @@ syncprov_op_response( Operation *op, SlapReply *rs )
|
|||||||
/* for each match in opc->smatches:
|
/* for each match in opc->smatches:
|
||||||
* send DELETE msg
|
* send DELETE msg
|
||||||
*/
|
*/
|
||||||
ldap_pvt_thread_mutex_lock( &si->si_ops_mutex );
|
|
||||||
for ( sm = opc->smatches; sm; sm=sm->sm_next ) {
|
for ( sm = opc->smatches; sm; sm=sm->sm_next ) {
|
||||||
if ( sm->sm_op->s_op->o_abandon )
|
if ( sm->sm_op->s_op->o_abandon )
|
||||||
continue;
|
continue;
|
||||||
syncprov_sendresp( op, opc, sm->sm_op, NULL,
|
syncprov_sendresp( op, opc, sm->sm_op, NULL,
|
||||||
LDAP_SYNC_DELETE, 1 );
|
LDAP_SYNC_DELETE, 1 );
|
||||||
}
|
}
|
||||||
ldap_pvt_thread_mutex_unlock( &si->si_ops_mutex );
|
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@ -1470,7 +1556,6 @@ syncprov_op_mod( Operation *op, SlapReply *rs )
|
|||||||
if (( si->si_ops || si->si_logs ) && op->o_tag != LDAP_REQ_ADD )
|
if (( si->si_ops || si->si_logs ) && op->o_tag != LDAP_REQ_ADD )
|
||||||
syncprov_matchops( op, opc, 1 );
|
syncprov_matchops( op, opc, 1 );
|
||||||
|
|
||||||
|
|
||||||
return SLAP_CB_CONTINUE;
|
return SLAP_CB_CONTINUE;
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -1633,41 +1718,9 @@ syncprov_search_response( Operation *op, SlapReply *rs )
|
|||||||
&cookie, 1, NULL, 0 );
|
&cookie, 1, NULL, 0 );
|
||||||
/* Flush any queued persist messages */
|
/* Flush any queued persist messages */
|
||||||
if ( ss->ss_so->s_res ) {
|
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 );
|
ldap_pvt_thread_mutex_lock( &ss->ss_so->s_mutex );
|
||||||
locked = 1;
|
locked = 1;
|
||||||
for (sr = ss->ss_so->s_res; sr; sr=srnext) {
|
syncprov_qplay( op, on, ss->ss_so );
|
||||||
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;
|
|
||||||
}
|
}
|
||||||
|
|
||||||
/* Turn off the refreshing flag */
|
/* Turn off the refreshing flag */
|
||||||
|
Loading…
Reference in New Issue
Block a user