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;
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 )
{
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;
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;
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;
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;
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;
}
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;
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 ) {
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;
/* 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;
}
}
if (( si->si_ops || si->si_logs ) && op->o_tag != LDAP_REQ_ADD )
syncprov_matchops( op, opc, 1 );
-
return SLAP_CB_CONTINUE;
}
&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 */
projects / openldap / commitdiff