/* syncprov.c - syncrepl provider */
/* This work is part of OpenLDAP Software <http://www.openldap.org/>.
*
- * Copyright 2004-2011 The OpenLDAP Foundation.
+ * Copyright 2004-2016 The OpenLDAP Foundation.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
typedef struct modtarget {
struct modinst *mt_mods;
struct modinst *mt_tail;
- Operation *mt_op;
+ struct berval mt_dn;
ldap_pvt_thread_mutex_t mt_mutex;
} modtarget;
+/* All the info of a psearch result that's shared between
+ * multiple queues
+ */
+typedef struct resinfo {
+ struct syncres *ri_list;
+ Entry *ri_e;
+ struct berval ri_dn;
+ struct berval ri_ndn;
+ struct berval ri_uuid;
+ struct berval ri_csn;
+ struct berval ri_cookie;
+ char ri_isref;
+ ldap_pvt_thread_mutex_t ri_mutex;
+} resinfo;
+
/* A queued result of a persistent search */
typedef struct syncres {
- struct syncres *s_next;
- Entry *s_e;
- struct berval s_dn;
- struct berval s_ndn;
- struct berval s_uuid;
- struct berval s_csn;
+ struct syncres *s_next; /* list of results on this psearch queue */
+ struct syncres *s_rilist; /* list of psearches using this result */
+ resinfo *s_info;
char s_mode;
- char s_isreference;
} syncres;
/* Record of a persistent search */
typedef struct syncops {
struct syncops *s_next;
+ struct syncprov_info_t *s_si;
struct berval s_base; /* ndn of search base */
ID s_eid; /* entryID of search base */
Operation *s_op; /* search op */
short osid; /* sid of op csn */
short rsid; /* sid of relay */
short sreference; /* Is the entry a reference? */
+ syncres ssres;
} opcookie;
-typedef struct mutexint {
- ldap_pvt_thread_mutex_t mi_mutex;
- int mi_int;
-} mutexint;
-
typedef struct fbase_cookie {
struct berval *fdn; /* DN of a modified entry, for scope testing */
syncops *fss; /* persistent search we're testing against */
BerElementBuffer berbuf;
BerElement *ber = (BerElement *)&berbuf;
-
+ LDAPControl *cp;
+ struct berval bv;
struct berval entryuuid_bv = BER_BVNULL;
ber_init2( ber, 0, LBER_USE_DER );
ber_set_option( ber, LBER_OPT_BER_MEMCTX, &op->o_tmpmemctx );
- ctrls[num_ctrls] = op->o_tmpalloc( sizeof ( LDAPControl ), op->o_tmpmemctx );
-
for ( a = e->e_attrs; a != NULL; a = a->a_next ) {
AttributeDescription *desc = a->a_desc;
if ( desc == slap_schema.si_ad_entryUUID ) {
entry_sync_state, &entryuuid_bv );
}
- ctrls[num_ctrls]->ldctl_oid = LDAP_CONTROL_SYNC_STATE;
- ctrls[num_ctrls]->ldctl_iscritical = (op->o_sync == SLAP_CONTROL_CRITICAL);
- ret = ber_flatten2( ber, &ctrls[num_ctrls]->ldctl_value, 1 );
-
+ ret = ber_flatten2( ber, &bv, 0 );
+ if ( ret == 0 ) {
+ cp = op->o_tmpalloc( sizeof( LDAPControl ) + bv.bv_len, op->o_tmpmemctx );
+ cp->ldctl_oid = LDAP_CONTROL_SYNC_STATE;
+ cp->ldctl_iscritical = (op->o_sync == SLAP_CONTROL_CRITICAL);
+ cp->ldctl_value.bv_val = (char *)&cp[1];
+ cp->ldctl_value.bv_len = bv.bv_len;
+ AC_MEMCPY( cp->ldctl_value.bv_val, bv.bv_val, bv.bv_len );
+ ctrls[num_ctrls] = cp;
+ }
ber_free_buf( ber );
if ( ret < 0 ) {
int ret;
BerElementBuffer berbuf;
BerElement *ber = (BerElement *)&berbuf;
+ LDAPControl *cp;
+ struct berval bv;
ber_init2( ber, NULL, LBER_USE_DER );
ber_set_option( ber, LBER_OPT_BER_MEMCTX, &op->o_tmpmemctx );
- ctrls[num_ctrls] = op->o_tmpalloc( sizeof ( LDAPControl ), op->o_tmpmemctx );
-
ber_printf( ber, "{" );
if ( send_cookie && cookie ) {
ber_printf( ber, "O", cookie );
}
ber_printf( ber, "N}" );
- ctrls[num_ctrls]->ldctl_oid = LDAP_CONTROL_SYNC_DONE;
- ctrls[num_ctrls]->ldctl_iscritical = (op->o_sync == SLAP_CONTROL_CRITICAL);
- ret = ber_flatten2( ber, &ctrls[num_ctrls]->ldctl_value, 1 );
+ ret = ber_flatten2( ber, &bv, 0 );
+ if ( ret == 0 ) {
+ cp = op->o_tmpalloc( sizeof( LDAPControl ) + bv.bv_len, op->o_tmpmemctx );
+ cp->ldctl_oid = LDAP_CONTROL_SYNC_DONE;
+ cp->ldctl_iscritical = (op->o_sync == SLAP_CONTROL_CRITICAL);
+ cp->ldctl_value.bv_val = (char *)&cp[1];
+ cp->ldctl_value.bv_len = bv.bv_len;
+ AC_MEMCPY( cp->ldctl_value.bv_val, bv.bv_val, bv.bv_len );
+ ctrls[num_ctrls] = cp;
+ }
ber_free_buf( ber );
int rc;
m1 = c1; m2 = c2;
- rc = m1->mt_op->o_req_ndn.bv_len - m2->mt_op->o_req_ndn.bv_len;
+ rc = m1->mt_dn.bv_len - m2->mt_dn.bv_len;
if ( rc ) return rc;
- return ber_bvcmp( &m1->mt_op->o_req_ndn, &m2->mt_op->o_req_ndn );
+ return ber_bvcmp( &m1->mt_dn, &m2->mt_dn );
}
/* syncprov_findbase:
fop.o_hdr = op->o_hdr;
fop.o_time = op->o_time;
fop.o_tincr = op->o_tincr;
+ fop.o_extra = op->o_extra;
cb.sc_response = findbase_cb;
cb.sc_private = fc;
fop.o_sync_mode = 0; /* turn off sync mode */
fop.o_managedsait = SLAP_CONTROL_CRITICAL;
fop.o_callback = &cb;
- LDAP_SLIST_INIT( &fop.o_extra );
fop.o_tag = LDAP_REQ_SEARCH;
fop.ors_scope = LDAP_SCOPE_BASE;
fop.ors_limit = NULL;
return rc;
}
-/* Should find a place to cache these */
-static mutexint *get_mutexint()
-{
- mutexint *mi = ch_malloc( sizeof( mutexint ));
- ldap_pvt_thread_mutex_init( &mi->mi_mutex );
- mi->mi_int = 1;
- return mi;
-}
-
-static void inc_mutexint( mutexint *mi )
-{
- ldap_pvt_thread_mutex_lock( &mi->mi_mutex );
- mi->mi_int++;
- ldap_pvt_thread_mutex_unlock( &mi->mi_mutex );
-}
-
-/* return resulting counter */
-static int dec_mutexint( mutexint *mi )
+static void free_resinfo( syncres *sr )
{
- int i;
- ldap_pvt_thread_mutex_lock( &mi->mi_mutex );
- i = --mi->mi_int;
- ldap_pvt_thread_mutex_unlock( &mi->mi_mutex );
- if ( !i ) {
- ldap_pvt_thread_mutex_destroy( &mi->mi_mutex );
- ch_free( mi );
- }
- return i;
+ syncres **st;
+ int freeit = 0;
+ ldap_pvt_thread_mutex_lock( &sr->s_info->ri_mutex );
+ for (st = &sr->s_info->ri_list; *st; st = &(*st)->s_rilist) {
+ if (*st == sr) {
+ *st = sr->s_rilist;
+ break;
+ }
+ }
+ if ( !sr->s_info->ri_list )
+ freeit = 1;
+ ldap_pvt_thread_mutex_unlock( &sr->s_info->ri_mutex );
+ if ( freeit ) {
+ ldap_pvt_thread_mutex_destroy( &sr->s_info->ri_mutex );
+ if ( sr->s_info->ri_e )
+ entry_free( sr->s_info->ri_e );
+ if ( !BER_BVISNULL( &sr->s_info->ri_cookie ))
+ ch_free( sr->s_info->ri_cookie.bv_val );
+ ch_free( sr->s_info );
+ }
}
-static void
-syncprov_free_syncop( syncops *so )
+static int
+syncprov_free_syncop( syncops *so, int unlink )
{
syncres *sr, *srnext;
GroupAssertion *ga, *gnext;
/* already being freed, or still in use */
if ( !so->s_inuse || --so->s_inuse > 0 ) {
ldap_pvt_thread_mutex_unlock( &so->s_mutex );
- return;
+ return 0;
}
ldap_pvt_thread_mutex_unlock( &so->s_mutex );
+ if ( unlink ) {
+ syncops **sop;
+ ldap_pvt_thread_mutex_lock( &so->s_si->si_ops_mutex );
+ for ( sop = &so->s_si->si_ops; *sop; sop = &(*sop)->s_next ) {
+ if ( *sop == so ) {
+ *sop = so->s_next;
+ break;
+ }
+ }
+ ldap_pvt_thread_mutex_unlock( &so->s_si->si_ops_mutex );
+ }
if ( so->s_flags & PS_IS_DETACHED ) {
filter_free( so->s_op->ors_filter );
for ( ga = so->s_op->o_groups; ga; ga=gnext ) {
ch_free( so->s_base.bv_val );
for ( sr=so->s_res; sr; sr=srnext ) {
srnext = sr->s_next;
- if ( sr->s_e ) {
- if ( !dec_mutexint( sr->s_e->e_private )) {
- sr->s_e->e_private = NULL;
- entry_free( sr->s_e );
- }
- }
+ free_resinfo( sr );
ch_free( sr );
}
ldap_pvt_thread_mutex_destroy( &so->s_mutex );
ch_free( so );
+ return 1;
}
/* Send a persistent search response */
static int
-syncprov_sendresp( Operation *op, opcookie *opc, syncops *so, int mode )
+syncprov_sendresp( Operation *op, resinfo *ri, syncops *so, int mode )
{
SlapReply rs = { REP_SEARCH };
- LDAPControl *ctrls[2];
- struct berval cookie = BER_BVNULL, csns[2];
+ struct berval cookie, csns[2];
Entry e_uuid = {0};
Attribute a_uuid = {0};
if ( so->s_op->o_abandon )
return SLAPD_ABANDON;
- ctrls[1] = NULL;
- if ( !BER_BVISNULL( &opc->sctxcsn )) {
- csns[0] = opc->sctxcsn;
- BER_BVZERO( &csns[1] );
- slap_compose_sync_cookie( op, &cookie, csns, so->s_rid, slap_serverID ? slap_serverID : -1 );
- }
+ rs.sr_ctrls = op->o_tmpalloc( sizeof(LDAPControl *)*2, op->o_tmpmemctx );
+ rs.sr_ctrls[1] = NULL;
+ rs.sr_flags = REP_CTRLS_MUSTBEFREED;
+ csns[0] = ri->ri_csn;
+ BER_BVZERO( &csns[1] );
+ slap_compose_sync_cookie( op, &cookie, csns, so->s_rid, slap_serverID ? slap_serverID : -1 );
#ifdef LDAP_DEBUG
if ( so->s_sid > 0 ) {
Debug( LDAP_DEBUG_SYNC, "syncprov_sendresp: to=%03x, cookie=%s\n",
- so->s_sid, cookie.bv_val ? cookie.bv_val : "", 0 );
+ so->s_sid, cookie.bv_val, 0 );
} else {
Debug( LDAP_DEBUG_SYNC, "syncprov_sendresp: cookie=%s\n",
- cookie.bv_val ? cookie.bv_val : "", 0, 0 );
+ cookie.bv_val, 0, 0 );
}
#endif
e_uuid.e_attrs = &a_uuid;
a_uuid.a_desc = slap_schema.si_ad_entryUUID;
- a_uuid.a_nvals = &opc->suuid;
+ a_uuid.a_nvals = &ri->ri_uuid;
rs.sr_err = syncprov_state_ctrl( op, &rs, &e_uuid,
- mode, ctrls, 0, 1, &cookie );
- if ( !BER_BVISNULL( &cookie )) {
- op->o_tmpfree( cookie.bv_val, op->o_tmpmemctx );
- }
+ mode, rs.sr_ctrls, 0, 1, &cookie );
+ op->o_tmpfree( cookie.bv_val, op->o_tmpmemctx );
- rs.sr_ctrls = ctrls;
rs.sr_entry = &e_uuid;
if ( mode == LDAP_SYNC_ADD || mode == LDAP_SYNC_MODIFY ) {
- e_uuid = *opc->se;
+ e_uuid = *ri->ri_e;
e_uuid.e_private = NULL;
}
switch( mode ) {
case LDAP_SYNC_ADD:
- if ( opc->sreference && so->s_op->o_managedsait <= SLAP_CONTROL_IGNORED ) {
+ if ( ri->ri_isref && so->s_op->o_managedsait <= SLAP_CONTROL_IGNORED ) {
rs.sr_ref = get_entry_referrals( op, rs.sr_entry );
rs.sr_err = send_search_reference( op, &rs );
ber_bvarray_free( rs.sr_ref );
break;
case LDAP_SYNC_DELETE:
e_uuid.e_attrs = NULL;
- e_uuid.e_name = opc->sdn;
- e_uuid.e_nname = opc->sndn;
- if ( opc->sreference && so->s_op->o_managedsait <= SLAP_CONTROL_IGNORED ) {
+ e_uuid.e_name = ri->ri_dn;
+ e_uuid.e_nname = ri->ri_ndn;
+ if ( ri->ri_isref && so->s_op->o_managedsait <= SLAP_CONTROL_IGNORED ) {
struct berval bv = BER_BVNULL;
rs.sr_ref = &bv;
rs.sr_err = send_search_reference( op, &rs );
default:
assert(0);
}
- /* In case someone else freed it already? */
- if ( rs.sr_ctrls ) {
- int i;
- for ( i=0; rs.sr_ctrls[i]; i++ ) {
- if ( rs.sr_ctrls[i] == ctrls[0] ) {
- op->o_tmpfree( ctrls[0]->ldctl_value.bv_val, op->o_tmpmemctx );
- ctrls[0]->ldctl_value.bv_val = NULL;
- break;
- }
- }
- rs.sr_ctrls = NULL;
- }
-
return rs.sr_err;
}
{
slap_overinst *on = LDAP_SLIST_FIRST(&so->s_op->o_extra)->oe_key;
syncres *sr;
- opcookie opc;
int rc = 0;
- opc.son = on;
-
do {
ldap_pvt_thread_mutex_lock( &so->s_mutex );
sr = so->s_res;
- if ( sr )
- so->s_res = sr->s_next;
- if ( !so->s_res )
- so->s_restail = NULL;
/* Exit loop with mutex held */
- if ( !sr || so->s_op->o_abandon )
+ if ( !sr )
break;
+ so->s_res = sr->s_next;
+ if ( !so->s_res )
+ so->s_restail = NULL;
ldap_pvt_thread_mutex_unlock( &so->s_mutex );
- if ( sr->s_mode == LDAP_SYNC_NEW_COOKIE ) {
- SlapReply rs = { REP_INTERMEDIATE };
+ if ( !so->s_op->o_abandon ) {
- rc = syncprov_sendinfo( op, &rs, LDAP_TAG_SYNC_NEW_COOKIE,
- &sr->s_csn, 0, NULL, 0 );
- } else {
- 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;
- opc.se = sr->s_e;
+ if ( sr->s_mode == LDAP_SYNC_NEW_COOKIE ) {
+ SlapReply rs = { REP_INTERMEDIATE };
- rc = syncprov_sendresp( op, &opc, so, sr->s_mode );
-
- }
- if ( sr->s_e ) {
- if ( !dec_mutexint( sr->s_e->e_private )) {
- sr->s_e->e_private = NULL;
- entry_free ( sr->s_e );
+ rc = syncprov_sendinfo( op, &rs, LDAP_TAG_SYNC_NEW_COOKIE,
+ &sr->s_info->ri_cookie, 0, NULL, 0 );
+ } else {
+ rc = syncprov_sendresp( op, sr->s_info, so, sr->s_mode );
}
}
+ free_resinfo( sr );
ch_free( sr );
+ if ( so->s_op->o_abandon )
+ continue;
+
/* Exit loop with mutex held */
ldap_pvt_thread_mutex_lock( &so->s_mutex );
+ break;
- } while (0);
+ } while (1);
/* We now only send one change at a time, to prevent one
* psearch from hogging all the CPU. Resubmit this task if
rc = syncprov_qplay( op, so );
/* decrement use count... */
- syncprov_free_syncop( so );
+ syncprov_free_syncop( so, 1 );
return NULL;
}
syncprov_qresp( opcookie *opc, syncops *so, int mode )
{
syncres *sr;
+ resinfo *ri;
int srsize;
- struct berval cookie = opc->sctxcsn;
-
- if ( mode == LDAP_SYNC_NEW_COOKIE ) {
- syncprov_info_t *si = opc->son->on_bi.bi_private;
-
- slap_compose_sync_cookie( NULL, &cookie, si->si_ctxcsn,
- so->s_rid, slap_serverID ? slap_serverID : -1);
- }
+ struct berval csn = opc->sctxcsn;
- srsize = sizeof(syncres) + opc->suuid.bv_len + 1 +
- opc->sdn.bv_len + 1 + opc->sndn.bv_len + 1;
- if ( cookie.bv_len )
- srsize += cookie.bv_len + 1;
- sr = ch_malloc( srsize );
+ sr = ch_malloc( sizeof( syncres ));
sr->s_next = NULL;
- sr->s_e = opc->se;
- /* bump refcount on this entry */
- if ( opc->se )
- inc_mutexint( opc->se->e_private );
- 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 ) + 1;
- sr->s_ndn.bv_len = opc->sndn.bv_len;
- sr->s_uuid.bv_val = lutil_strcopy( sr->s_ndn.bv_val,
- opc->sndn.bv_val ) + 1;
- sr->s_uuid.bv_len = opc->suuid.bv_len;
- AC_MEMCPY( sr->s_uuid.bv_val, opc->suuid.bv_val, opc->suuid.bv_len );
- if ( cookie.bv_len ) {
- sr->s_csn.bv_val = sr->s_uuid.bv_val + sr->s_uuid.bv_len + 1;
- strcpy( sr->s_csn.bv_val, cookie.bv_val );
- } else {
- sr->s_csn.bv_val = NULL;
- }
- sr->s_csn.bv_len = cookie.bv_len;
+ if ( !opc->ssres.s_info ) {
+
+ srsize = sizeof( resinfo );
+ if ( csn.bv_len )
+ srsize += csn.bv_len + 1;
+
+ if ( opc->se ) {
+ Attribute *a;
+ ri = ch_malloc( srsize );
+ ri->ri_dn = opc->se->e_name;
+ ri->ri_ndn = opc->se->e_nname;
+ a = attr_find( opc->se->e_attrs, slap_schema.si_ad_entryUUID );
+ if ( a )
+ ri->ri_uuid = a->a_nvals[0];
+ else
+ ri->ri_uuid.bv_len = 0;
+ if ( csn.bv_len ) {
+ ri->ri_csn.bv_val = (char *)(ri + 1);
+ ri->ri_csn.bv_len = csn.bv_len;
+ memcpy( ri->ri_csn.bv_val, csn.bv_val, csn.bv_len );
+ ri->ri_csn.bv_val[csn.bv_len] = '\0';
+ } else {
+ ri->ri_csn.bv_val = NULL;
+ }
+ } else {
+ srsize += opc->suuid.bv_len +
+ opc->sdn.bv_len + 1 + opc->sndn.bv_len + 1;
+ ri = ch_malloc( srsize );
+ ri->ri_dn.bv_val = (char *)(ri + 1);
+ ri->ri_dn.bv_len = opc->sdn.bv_len;
+ ri->ri_ndn.bv_val = lutil_strcopy( ri->ri_dn.bv_val,
+ opc->sdn.bv_val ) + 1;
+ ri->ri_ndn.bv_len = opc->sndn.bv_len;
+ ri->ri_uuid.bv_val = lutil_strcopy( ri->ri_ndn.bv_val,
+ opc->sndn.bv_val ) + 1;
+ ri->ri_uuid.bv_len = opc->suuid.bv_len;
+ AC_MEMCPY( ri->ri_uuid.bv_val, opc->suuid.bv_val, opc->suuid.bv_len );
+ if ( csn.bv_len ) {
+ ri->ri_csn.bv_val = ri->ri_uuid.bv_val + ri->ri_uuid.bv_len;
+ memcpy( ri->ri_csn.bv_val, csn.bv_val, csn.bv_len );
+ ri->ri_csn.bv_val[csn.bv_len] = '\0';
+ } else {
+ ri->ri_csn.bv_val = NULL;
+ }
+ }
+ ri->ri_list = &opc->ssres;
+ ri->ri_e = opc->se;
+ ri->ri_csn.bv_len = csn.bv_len;
+ ri->ri_isref = opc->sreference;
+ BER_BVZERO( &ri->ri_cookie );
+ ldap_pvt_thread_mutex_init( &ri->ri_mutex );
+ opc->se = NULL;
+ opc->ssres.s_info = ri;
+ }
+ ri = opc->ssres.s_info;
+ sr->s_info = ri;
+ ldap_pvt_thread_mutex_lock( &ri->ri_mutex );
+ sr->s_rilist = ri->ri_list;
+ ri->ri_list = sr;
+ if ( mode == LDAP_SYNC_NEW_COOKIE && BER_BVISNULL( &ri->ri_cookie )) {
+ syncprov_info_t *si = opc->son->on_bi.bi_private;
- if ( mode == LDAP_SYNC_NEW_COOKIE && cookie.bv_val ) {
- ch_free( cookie.bv_val );
+ slap_compose_sync_cookie( NULL, &ri->ri_cookie, si->si_ctxcsn,
+ so->s_rid, slap_serverID ? slap_serverID : -1);
}
+ ldap_pvt_thread_mutex_unlock( &ri->ri_mutex );
ldap_pvt_thread_mutex_lock( &so->s_mutex );
if ( !so->s_res ) {
if ( lock )
ldap_pvt_thread_mutex_unlock( &so->s_op->o_conn->c_mutex );
}
- syncprov_free_syncop( so );
-
- return 0;
+ return syncprov_free_syncop( so, 0 );
}
static int
{
slap_callback *sc = op->o_callback;
op->o_callback = sc->sc_next;
- syncprov_drop_psearch( op->o_callback->sc_private, 0 );
+ syncprov_drop_psearch( sc->sc_private, 0 );
op->o_tmpfree( sc, op->o_tmpmemctx );
return 0;
}
{
slap_overinst *on = (slap_overinst *)op->o_bd->bd_info;
syncprov_info_t *si = on->on_bi.bi_private;
- syncops *so, *soprev;
+ syncops *so, **sop;
ldap_pvt_thread_mutex_lock( &si->si_ops_mutex );
- for ( so=si->si_ops, soprev = (syncops *)&si->si_ops; so;
- soprev=so, so=so->s_next ) {
+ for ( sop=&si->si_ops; (so = *sop); sop = &(*sop)->s_next ) {
if ( so->s_op->o_connid == op->o_connid &&
so->s_op->o_msgid == op->orn_msgid ) {
so->s_op->o_abandon = 1;
- soprev->s_next = so->s_next;
+ *sop = so->s_next;
break;
}
}
cb->sc_cleanup = syncprov_ab_cleanup;
cb->sc_next = op->o_callback;
cb->sc_private = so;
+ op->o_callback = cb;
return SLAP_CB_CONTINUE;
}
}
syncprov_info_t *si = on->on_bi.bi_private;
fbase_cookie fc;
- syncops *ss, *sprev, *snext;
+ syncops **pss;
Entry *e = NULL;
Attribute *a;
- int rc;
+ int rc, gonext;
struct berval newdn;
int freefdn = 0;
BackendDB *b0 = op->o_bd, db;
rc = overlay_entry_get_ov( op, fc.fdn, NULL, NULL, 0, &e, on );
/* If we're sending responses now, make a copy and unlock the DB */
if ( e && !saveit ) {
- if ( !opc->se ) {
+ if ( !opc->se )
opc->se = entry_dup( e );
- opc->se->e_private = get_mutexint();
- }
overlay_entry_release_ov( op, e, 0, on );
e = opc->se;
}
} else {
e = op->ora_e;
if ( !saveit ) {
- if ( !opc->se ) {
+ if ( !opc->se )
opc->se = entry_dup( e );
- opc->se->e_private = get_mutexint();
- }
e = opc->se;
}
}
}
ldap_pvt_thread_mutex_lock( &si->si_ops_mutex );
- for (ss = si->si_ops, sprev = (syncops *)&si->si_ops; ss;
- sprev = ss, ss=snext)
+ for (pss = &si->si_ops; *pss; pss = gonext ? &(*pss)->s_next : pss)
{
Operation op2;
Opheader oh;
syncmatches *sm;
int found = 0;
+ syncops *snext, *ss = *pss;
- snext = ss->s_next;
+ gonext = 1;
if ( ss->s_op->o_abandon )
continue;
SlapReply rs = {REP_RESULT};
send_ldap_error( ss->s_op, &rs, LDAP_SYNC_REFRESH_REQUIRED,
"search base has changed" );
- sprev->s_next = snext;
- syncprov_drop_psearch( ss, 1 );
- ss = sprev;
+ snext = ss->s_next;
+ if ( syncprov_drop_psearch( ss, 1 ) )
+ *pss = snext;
+ gonext = 0;
continue;
}
phase otherwise (ITS#6555) */
op2.ors_filter = ss->s_op->ors_filter->f_and->f_next;
}
- ldap_pvt_thread_mutex_unlock( &ss->s_mutex );
rc = test_filter( &op2, e, op2.ors_filter );
+ ldap_pvt_thread_mutex_unlock( &ss->s_mutex );
}
Debug( LDAP_DEBUG_TRACE, "syncprov_matchops: sid %03x fscope %d rc %d\n",
/* Decrement s_inuse, was incremented when called
* with saveit == TRUE
*/
- syncprov_free_syncop( ss );
+ snext = ss->s_next;
+ if ( syncprov_free_syncop( ss, 0 ) ) {
+ *pss = snext;
+ gonext = 0;
+ }
}
}
ldap_pvt_thread_mutex_unlock( &si->si_ops_mutex );
overlay_entry_release_ov( op, e, 0, on );
op->o_bd = b0;
}
- if ( opc->se && !saveit ) {
- if ( !dec_mutexint( opc->se->e_private )) {
- opc->se->e_private = NULL;
+ if ( !saveit ) {
+ if ( opc->ssres.s_info )
+ free_resinfo( &opc->ssres );
+ else if ( opc->se )
entry_free( opc->se );
- opc->se = NULL;
- }
}
if ( freefdn ) {
op->o_tmpfree( fc.fdn->bv_val, op->o_tmpmemctx );
for (sm = opc->smatches; sm; sm=snext) {
snext = sm->sm_next;
- syncprov_free_syncop( sm->sm_op );
+ syncprov_free_syncop( sm->sm_op, 1 );
op->o_tmpfree( sm, op->o_tmpmemctx );
}
/* Remove op from lock table */
mt = opc->smt;
if ( mt ) {
+ modinst *mi = (modinst *)(opc+1), **m2;
ldap_pvt_thread_mutex_lock( &mt->mt_mutex );
- mt->mt_mods = mt->mt_mods->mi_next;
+ for (m2 = &mt->mt_mods; ; m2 = &(*m2)->mi_next) {
+ if ( *m2 == mi ) {
+ *m2 = mi->mi_next;
+ if ( mt->mt_tail == mi )
+ mt->mt_tail = ( m2 == &mt->mt_mods ) ? NULL : (modinst *)m2;
+ break;
+ }
+ }
/* If there are more, promote the next one */
if ( mt->mt_mods ) {
- mt->mt_op = mt->mt_mods->mi_op;
ldap_pvt_thread_mutex_unlock( &mt->mt_mutex );
} else {
ldap_pvt_thread_mutex_unlock( &mt->mt_mutex );
avl_delete( &si->si_mods, mt, sp_avl_cmp );
ldap_pvt_thread_mutex_unlock( &si->si_mods_mutex );
ldap_pvt_thread_mutex_destroy( &mt->mt_mutex );
+ ch_free( mt->mt_dn.bv_val );
ch_free( mt );
}
}
} else {
ldap_pvt_thread_rdwr_wunlock( &si->si_csn_rwlock );
}
+ if ( csn_changed )
+ si->si_numops++;
goto leave;
}
- si->si_numops++;
+ if ( csn_changed )
+ si->si_numops++;
if ( si->si_chkops || si->si_chktime ) {
/* Never checkpoint adding the context entry,
* it will deadlock
mi->mi_op = op;
/* See if we're already modifying this entry... */
- mtdummy.mt_op = op;
+ mtdummy.mt_dn = op->o_req_ndn;
+retry:
ldap_pvt_thread_mutex_lock( &si->si_mods_mutex );
mt = avl_find( si->si_mods, &mtdummy, sp_avl_cmp );
if ( mt ) {
ldap_pvt_thread_mutex_lock( &mt->mt_mutex );
if ( mt->mt_mods == NULL ) {
/* Cannot reuse this mt, as another thread is about
- * to release it in syncprov_op_cleanup.
+ * to release it in syncprov_op_cleanup. Wait for them
+ * to finish; our own insert is required to succeed.
*/
ldap_pvt_thread_mutex_unlock( &mt->mt_mutex );
- mt = NULL;
+ ldap_pvt_thread_mutex_unlock( &si->si_mods_mutex );
+ ldap_pvt_thread_yield();
+ goto retry;
}
}
if ( mt ) {
- ldap_pvt_thread_mutex_unlock( &si->si_mods_mutex );
mt->mt_tail->mi_next = mi;
mt->mt_tail = mi;
+ ldap_pvt_thread_mutex_unlock( &si->si_mods_mutex );
/* wait for this op to get to head of list */
while ( mt->mt_mods != mi ) {
+ modinst *m2;
+ /* don't wait on other mods from the same thread */
+ for ( m2 = mt->mt_mods; m2; m2 = m2->mi_next ) {
+ if ( m2->mi_op->o_threadctx == op->o_threadctx ) {
+ break;
+ }
+ }
+ if ( m2 )
+ break;
+
ldap_pvt_thread_mutex_unlock( &mt->mt_mutex );
/* FIXME: if dynamic config can delete overlays or
* databases we'll have to check for cleanup here.
/* clean up if the caller is giving up */
if ( op->o_abandon ) {
- modinst *m2;
- for ( m2 = mt->mt_mods; m2->mi_next != mi;
- m2 = m2->mi_next );
- m2->mi_next = mi->mi_next;
- if ( mt->mt_tail == mi ) mt->mt_tail = m2;
+ modinst **m2;
+ slap_callback **sc;
+ for (m2 = &mt->mt_mods; ; m2 = &(*m2)->mi_next) {
+ if ( *m2 == mi ) {
+ *m2 = mi->mi_next;
+ if ( mt->mt_tail == mi )
+ mt->mt_tail = ( m2 == &mt->mt_mods ) ? NULL : (modinst *)m2;
+ break;
+ }
+ }
+ for (sc = &op->o_callback; ; sc = &(*sc)->sc_next) {
+ if ( *sc == cb ) {
+ *sc = cb->sc_next;
+ break;
+ }
+ }
op->o_tmpfree( cb, op->o_tmpmemctx );
ldap_pvt_thread_mutex_unlock( &mt->mt_mutex );
return SLAPD_ABANDON;
mt = ch_malloc( sizeof(modtarget) );
mt->mt_mods = mi;
mt->mt_tail = mi;
- mt->mt_op = mi->mi_op;
+ ber_dupbv( &mt->mt_dn, &mi->mi_op->o_req_ndn );
ldap_pvt_thread_mutex_init( &mt->mt_mutex );
avl_insert( &si->si_mods, mt, sp_avl_cmp, avl_dup_error );
ldap_pvt_thread_mutex_unlock( &si->si_mods_mutex );
int ss_flags;
} searchstate;
-static int
-syncprov_search_cleanup( Operation *op, SlapReply *rs )
-{
- if ( rs->sr_ctrls ) {
- op->o_tmpfree( rs->sr_ctrls[0], op->o_tmpmemctx );
- op->o_tmpfree( rs->sr_ctrls, op->o_tmpmemctx );
- rs->sr_ctrls = NULL;
- }
- return 0;
-}
-
typedef struct SyncOperationBuffer {
Operation sob_op;
Opheader sob_hdr;
slap_compose_sync_cookie( op, &cookie, a->a_nvals, srs->sr_state.rid, slap_serverID ? slap_serverID : -1 );
rs->sr_err = syncprov_state_ctrl( op, rs, rs->sr_entry,
LDAP_SYNC_ADD, rs->sr_ctrls, 0, 1, &cookie );
+ op->o_tmpfree( cookie.bv_val, op->o_tmpmemctx );
} else {
rs->sr_err = syncprov_state_ctrl( op, rs, rs->sr_entry,
LDAP_SYNC_ADD, rs->sr_ctrls, 0, 0, NULL );
rs->sr_ctrls = op->o_tmpalloc( sizeof(LDAPControl *)*2,
op->o_tmpmemctx );
rs->sr_ctrls[1] = NULL;
+ rs->sr_flags |= REP_CTRLS_MUSTBEFREED;
rs->sr_err = syncprov_done_ctrl( op, rs, rs->sr_ctrls,
0, 1, &cookie, ( ss->ss_flags & SS_PRESENT ) ? LDAP_SYNC_REFRESH_PRESENTS :
LDAP_SYNC_REFRESH_DELETES );
}
sop = ch_malloc( sizeof( syncops ));
*sop = so;
- ldap_pvt_thread_mutex_init( &sop->s_mutex );
sop->s_rid = srs->sr_state.rid;
sop->s_sid = srs->sr_state.sid;
- sop->s_inuse = 1;
+ /* set refcount=2 to prevent being freed out from under us
+ * by abandons that occur while we're running here
+ */
+ sop->s_inuse = 2;
ldap_pvt_thread_mutex_lock( &si->si_ops_mutex );
while ( si->si_active ) {
* active mods gets added to it.
*/
ldap_pvt_thread_mutex_unlock( &si->si_ops_mutex );
- if ( slapd_shutdown )
+ if ( slapd_shutdown ) {
+ ch_free( sop );
return SLAPD_ABANDON;
+ }
if ( !ldap_pvt_thread_pool_pausecheck( &connection_pool ))
ldap_pvt_thread_yield();
ldap_pvt_thread_mutex_lock( &si->si_ops_mutex );
}
+ if ( op->o_abandon ) {
+ ldap_pvt_thread_mutex_unlock( &si->si_ops_mutex );
+ ch_free( sop );
+ return SLAPD_ABANDON;
+ }
+ ldap_pvt_thread_mutex_init( &sop->s_mutex );
sop->s_next = si->si_ops;
+ sop->s_si = si;
si->si_ops = sop;
ldap_pvt_thread_mutex_unlock( &si->si_ops_mutex );
}
- /* snapshot the ctxcsn */
+ /* snapshot the ctxcsn
+ * Note: this must not be done before the psearch setup. (ITS#8365)
+ */
ldap_pvt_thread_rdwr_rlock( &si->si_csn_rwlock );
numcsns = si->si_numcsns;
if ( numcsns ) {
}
dirty = si->si_dirty;
ldap_pvt_thread_rdwr_runlock( &si->si_csn_rwlock );
-
+
/* If we have a cookie, handle the PRESENT lookups */
if ( srs->sr_state.ctxcsn ) {
sessionlog *sl;
int i, j;
- /* If we don't have any CSN of our own yet, pretend nothing
- * has changed.
+ /* If we don't have any CSN of our own yet, bail out.
*/
- if ( !numcsns )
- goto no_change;
+ if ( !numcsns ) {
+ rs->sr_err = LDAP_UNWILLING_TO_PERFORM;
+ rs->sr_text = "consumer has state info but provider doesn't!";
+ goto bailout;
+ }
if ( !si->si_nopres )
do_present = SS_PRESENT;
/* If nothing has changed, shortcut it */
if ( !changed && !dirty ) {
do_present = 0;
-no_change: if ( !(op->o_sync_mode & SLAP_SYNC_PERSIST) ) {
+no_change: if ( !(op->o_sync_mode & SLAP_SYNC_PERSIST) ) {
LDAPControl *ctrls[2];
ctrls[0] = NULL;
do_play = 1;
break;
}
- /* SID present and new enough */
- if ( minsid == sl->sl_sids[i]
- && ber_bvcmp( &mincsn, &sl->sl_mincsn[i] ) >= 0 ) {
- do_play = 1;
+ /* SID present */
+ if ( minsid == sl->sl_sids[i] ) {
+ /* new enough? */
+ if ( ber_bvcmp( &mincsn, &sl->sl_mincsn[i] ) >= 0 )
+ do_play = 1;
break;
}
}
}
}
} else {
+ /* The consumer knows nothing, we know nothing. OK. */
+ if (!numcsns)
+ goto no_change;
/* No consumer state, assume something has changed */
changed = SS_CHANGED;
}
* for persistent search evaluation
*/
if ( sop ) {
- sop->s_filterstr= op->ors_filterstr;
+ ldap_pvt_thread_mutex_lock( &sop->s_mutex );
+ sop->s_filterstr = op->ors_filterstr;
+ /* correct the refcount that was set to 2 before */
+ sop->s_inuse--;
}
/* If something changed, find the changes */
#endif
ber_dupbv_x( &fava->f_ava->aa_value, &mincsn, op->o_tmpmemctx );
fava->f_next = op->ors_filter;
- if ( sop )
- ldap_pvt_thread_mutex_lock( &sop->s_mutex );
op->ors_filter = fand;
filter2bv_x( op, op->ors_filter, &op->ors_filterstr );
if ( sop ) {
sop->s_flags |= PS_FIX_FILTER;
- ldap_pvt_thread_mutex_unlock( &sop->s_mutex );
}
}
+ if ( sop ) {
+ ldap_pvt_thread_mutex_unlock( &sop->s_mutex );
+ }
/* Let our callback add needed info to returned entries */
cb = op->o_tmpcalloc(1, sizeof(slap_callback)+sizeof(searchstate), op->o_tmpmemctx);
ss->ss_numcsns = numcsns;
ss->ss_sids = sids;
cb->sc_response = syncprov_search_response;
- cb->sc_cleanup = syncprov_search_cleanup;
cb->sc_private = ss;
cb->sc_next = op->o_callback;
op->o_callback = cb;
}
overlay_entry_release_ov( op, e, 0, on );
if ( si->si_ctxcsn && !SLAP_DBCLEAN( be )) {
+ op->o_tag = LDAP_REQ_SEARCH;
op->o_req_dn = be->be_suffix[0];
op->o_req_ndn = be->be_nsuffix[0];
op->ors_scope = LDAP_SCOPE_SUBTREE;
char csnbuf[ LDAP_PVT_CSNSTR_BUFSIZE ];
struct berval csn;
- if ( SLAP_SYNC_SHADOW( op->o_bd )) {
+ if ( slap_serverID || SLAP_SYNC_SHADOW( op->o_bd )) {
/* If we're also a consumer, then don't generate anything.
* Wait for our provider to send it to us, or for a local
* modify if we have multimaster.
#ifdef SLAP_CONFIG_DELETE
if ( !slapd_shutdown ) {
+ ldap_pvt_thread_mutex_lock( &si->si_ops_mutex );
for ( so=si->si_ops, sonext=so; so; so=sonext ) {
SlapReply rs = {REP_RESULT};
rs.sr_err = LDAP_UNAVAILABLE;
syncprov_drop_psearch( so, 0);
}
si->si_ops=NULL;
+ ldap_pvt_thread_mutex_unlock( &si->si_ops_mutex );
}
overlay_unregister_control( be, LDAP_CONTROL_SYNC );
#endif /* SLAP_CONFIG_DELETE */
projects / openldap / blobdiff