#define PS_IS_DETACHED 0x02
#define PS_WROTE_BASE 0x04
#define PS_FIND_BASE 0x08
+#define PS_FIX_FILTER 0x10
int s_inuse; /* reference count */
struct syncres *s_res;
syncres *sr;
Entry *e;
opcookie opc;
- int rc;
+ int rc = 0;
opc.son = on;
op->o_bd->bd_info = (BackendInfo *)on->on_info;
op->o_private = NULL;
op->o_callback = NULL;
- syncprov_qplay( op, on, so );
+ (void)syncprov_qplay( op, on, so );
/* decrement use count... */
syncprov_free_syncop( so );
return NULL;
}
+/* Start the task to play back queued psearch responses */
+static void
+syncprov_qstart( syncops *so )
+{
+ int wake=0;
+ ldap_pvt_thread_mutex_lock( &slapd_rq.rq_mutex );
+ if ( !so->s_qtask ) {
+ so->s_qtask = ldap_pvt_runqueue_insert( &slapd_rq, RUNQ_INTERVAL,
+ syncprov_qtask, so, "syncprov_qtask",
+ so->s_op->o_conn->c_peer_name.bv_val );
+ ++so->s_inuse;
+ wake = 1;
+ } else {
+ if (!ldap_pvt_runqueue_isrunning( &slapd_rq, so->s_qtask ) &&
+ !so->s_qtask->next_sched.tv_sec ) {
+ so->s_qtask->interval.tv_sec = 0;
+ ldap_pvt_runqueue_resched( &slapd_rq, so->s_qtask, 0 );
+ so->s_qtask->interval.tv_sec = RUNQ_INTERVAL;
+ ++so->s_inuse;
+ wake = 1;
+ }
+ }
+ ldap_pvt_thread_mutex_unlock( &slapd_rq.rq_mutex );
+ if ( wake )
+ slap_wake_listener();
+}
+
/* Queue a persistent search response */
static int
syncprov_qresp( opcookie *opc, syncops *so, int mode )
so->s_flags |= PS_FIND_BASE;
}
if ( so->s_flags & PS_IS_DETACHED ) {
- ldap_pvt_thread_mutex_lock( &slapd_rq.rq_mutex );
- if ( !so->s_qtask ) {
- so->s_qtask = ldap_pvt_runqueue_insert( &slapd_rq, RUNQ_INTERVAL,
- syncprov_qtask, so, "syncprov_qtask",
- so->s_op->o_conn->c_peer_name.bv_val );
- ++so->s_inuse;
- } else {
- if (!ldap_pvt_runqueue_isrunning( &slapd_rq, so->s_qtask ) &&
- !so->s_qtask->next_sched.tv_sec ) {
- so->s_qtask->interval.tv_sec = 0;
- ldap_pvt_runqueue_resched( &slapd_rq, so->s_qtask, 0 );
- so->s_qtask->interval.tv_sec = RUNQ_INTERVAL;
- ++so->s_inuse;
- }
- }
- ldap_pvt_thread_mutex_unlock( &slapd_rq.rq_mutex );
+ syncprov_qstart( so );
}
ldap_pvt_thread_mutex_unlock( &so->s_mutex );
return LDAP_SUCCESS;
int rc;
struct berval newdn;
int freefdn = 0;
+ BackendDB *b0 = op->o_bd, db;
fc.fdn = &op->o_req_ndn;
/* compute new DN */
freefdn = 1;
}
if ( op->o_tag != LDAP_REQ_ADD ) {
+ if ( !SLAP_ISOVERLAY( op->o_bd )) {
+ db = *op->o_bd;
+ op->o_bd = &db;
+ }
op->o_bd->bd_info = (BackendInfo *)on->on_info;
rc = be_entry_get_rw( op, fc.fdn, NULL, NULL, 0, &e );
/* If we're sending responses now, make a copy and unlock the DB */
e = e2;
}
op->o_bd->bd_info = (BackendInfo *)on;
- if ( rc ) return;
+ if ( rc ) {
+ op->o_bd = b0;
+ return;
+ }
} else {
e = op->ora_e;
}
if ( freefdn ) {
op->o_tmpfree( fc.fdn->bv_val, op->o_tmpmemctx );
}
+ op->o_bd = b0;
}
static int
struct berval bv[2];
slap_callback cb = {0};
- mod.sml_values = bv;
- bv[1].bv_val = NULL;
- bv[0] = si->si_ctxcsn;
+ /* If ctxcsn is empty, delete it */
+ if ( BER_BVISEMPTY( &si->si_ctxcsn )) {
+ mod.sml_values = NULL;
+ } else {
+ mod.sml_values = bv;
+ bv[1].bv_val = NULL;
+ bv[0] = si->si_ctxcsn;
+ }
mod.sml_nvalues = NULL;
mod.sml_desc = slap_schema.si_ad_contextCSN;
mod.sml_op = LDAP_MOD_REPLACE;
}
static void
-syncprov_add_slog( Operation *op, struct berval *csn )
+syncprov_add_slog( Operation *op )
{
opcookie *opc = op->o_callback->sc_private;
slap_overinst *on = opc->son;
{
/* Allocate a record. UUIDs are not NUL-terminated. */
se = ch_malloc( sizeof( slog_entry ) + opc->suuid.bv_len +
- csn->bv_len + 1 );
+ op->o_csn.bv_len + 1 );
se->se_next = NULL;
se->se_tag = op->o_tag;
- se->se_uuid.bv_val = (char *)(se+1);
+ se->se_uuid.bv_val = (char *)(&se[1]);
AC_MEMCPY( se->se_uuid.bv_val, opc->suuid.bv_val, opc->suuid.bv_len );
se->se_uuid.bv_len = opc->suuid.bv_len;
se->se_csn.bv_val = se->se_uuid.bv_val + opc->suuid.bv_len;
- AC_MEMCPY( se->se_csn.bv_val, csn->bv_val, csn->bv_len );
- se->se_csn.bv_val[csn->bv_len] = '\0';
- se->se_csn.bv_len = csn->bv_len;
+ AC_MEMCPY( se->se_csn.bv_val, op->o_csn.bv_val, op->o_csn.bv_len );
+ se->se_csn.bv_val[op->o_csn.bv_len] = '\0';
+ se->se_csn.bv_len = op->o_csn.bv_len;
ldap_pvt_thread_mutex_lock( &sl->sl_mutex );
if ( sl->sl_head ) {
sl->sl_mincsn.bv_len = se->se_csn.bv_len;
ch_free( se );
sl->sl_num--;
- if ( !sl->sl_head ) {
- sl->sl_tail = NULL;
- }
}
ldap_pvt_thread_mutex_unlock( &sl->sl_mutex );
}
/* enter with sl->sl_mutex locked, release before returning */
static void
syncprov_playlog( Operation *op, SlapReply *rs, sessionlog *sl,
- struct berval *oldcsn, struct berval *ctxcsn )
+ sync_control *srs, struct berval *ctxcsn )
{
slap_overinst *on = (slap_overinst *)op->o_bd->bd_info;
slog_entry *se;
int i, j, ndel, num, nmods, mmods;
+ char cbuf[LDAP_LUTIL_CSNSTR_BUFSIZE];
BerVarray uuids;
+ struct berval delcsn;
if ( !sl->sl_num ) {
ldap_pvt_thread_mutex_unlock( &sl->sl_mutex );
uuids = op->o_tmpalloc( (num+1) * sizeof( struct berval ) +
num * UUID_LEN, op->o_tmpmemctx );
-
uuids[0].bv_val = (char *)(uuids + num + 1);
+ delcsn.bv_len = 0;
+ delcsn.bv_val = cbuf;
+
/* Make a copy of the relevant UUIDs. Put the Deletes up front
* and everything else at the end. Do this first so we can
* unlock the list mutex.
*/
for ( se=sl->sl_head; se; se=se->se_next ) {
- if ( ber_bvcmp( &se->se_csn, oldcsn ) < 0 ) continue;
+ if ( ber_bvcmp( &se->se_csn, &srs->sr_state.ctxcsn ) <= 0 ) continue;
if ( ber_bvcmp( &se->se_csn, ctxcsn ) > 0 ) break;
if ( se->se_tag == LDAP_REQ_DELETE ) {
j = i;
i++;
+ AC_MEMCPY( cbuf, se->se_csn.bv_val, se->se_csn.bv_len );
+ delcsn.bv_len = se->se_csn.bv_len;
+ delcsn.bv_val[delcsn.bv_len] = '\0';
} else {
nmods++;
j = num - nmods;
fop.o_bd->bd_info = (BackendInfo *)on;
}
if ( ndel ) {
+ struct berval cookie;
+
+ slap_compose_sync_cookie( op, &cookie, &delcsn, srs->sr_state.rid );
uuids[ndel].bv_val = NULL;
- syncprov_sendinfo( op, rs, LDAP_TAG_SYNC_ID_SET, NULL, 0, uuids, 1 );
+ syncprov_sendinfo( op, rs, LDAP_TAG_SYNC_ID_SET, &cookie, 0, uuids, 1 );
+ op->o_tmpfree( cookie.bv_val, op->o_tmpmemctx );
}
+ op->o_tmpfree( uuids, op->o_tmpmemctx );
}
static int
if ( rs->sr_err == LDAP_SUCCESS )
{
- struct berval maxcsn = BER_BVNULL, curcsn = BER_BVNULL;
+ struct berval maxcsn = BER_BVNULL;
char cbuf[LDAP_LUTIL_CSNSTR_BUFSIZE];
/* Update our context CSN */
cbuf[0] = '\0';
ldap_pvt_thread_mutex_lock( &si->si_csn_mutex );
- slap_get_commit_csn( op, &maxcsn, &curcsn );
+ slap_get_commit_csn( op, &maxcsn );
if ( !BER_BVISNULL( &maxcsn ) ) {
strcpy( cbuf, maxcsn.bv_val );
if ( ber_bvcmp( &maxcsn, &si->si_ctxcsn ) > 0 ) {
/* Add any log records */
if ( si->si_logs && op->o_tag != LDAP_REQ_ADD ) {
- syncprov_add_slog( op, &curcsn );
+ syncprov_add_slog( op );
}
}
op2->ors_filterstr.bv_val = ptr;
strcpy( ptr, so->s_filterstr.bv_val );
op2->ors_filterstr.bv_len = so->s_filterstr.bv_len;
- op2->ors_filter = filter_dup( op->ors_filter, NULL );
+
+ /* Skip the AND/GE clause that we stuck on in front */
+ if ( so->s_flags & PS_FIX_FILTER ) {
+ op2->ors_filter = op->ors_filter->f_and->f_next;
+ so->s_flags ^= PS_FIX_FILTER;
+ } else {
+ op2->ors_filter = op->ors_filter;
+ }
+ op2->ors_filter = filter_dup( op2->ors_filter, NULL );
so->s_op = op2;
/* Copy any cached group ACLs individually */
LDAP_STAILQ_INSERT_TAIL( &op->o_conn->c_ops, op2, o_next );
so->s_flags |= PS_IS_DETACHED;
ldap_pvt_thread_mutex_unlock( &op->o_conn->c_mutex );
+
+ /* Prevent anyone else from trying to send a result for this op */
+ op->o_abandon = 1;
}
static int
a = attr_find( rs->sr_operational_attrs, slap_schema.si_ad_entryCSN );
}
if ( a ) {
- /* Make sure entry is less than the snaphot'd contextCSN */
- if ( ber_bvcmp( &a->a_nvals[0], &ss->ss_ctxcsn ) > 0 )
+ /* Make sure entry is less than the snapshot'd contextCSN */
+ if ( ber_bvcmp( &a->a_nvals[0], &ss->ss_ctxcsn ) > 0 ) {
+ Debug( LDAP_DEBUG_SYNC, "Entry %s CSN %s greater than snapshot %s\n",
+ rs->sr_entry->e_name.bv_val,
+ a->a_nvals[0].bv_val,
+ ss->ss_ctxcsn.bv_val );
return LDAP_SUCCESS;
+ }
/* Don't send the ctx entry twice */
if ( !BER_BVISNULL( &srs->sr_state.ctxcsn ) &&
- bvmatch( &a->a_nvals[0], &srs->sr_state.ctxcsn ) )
+ bvmatch( &a->a_nvals[0], &srs->sr_state.ctxcsn ) ) {
+ Debug( LDAP_DEBUG_SYNC, "Entry %s CSN %s matches ctx %s\n",
+ rs->sr_entry->e_name.bv_val,
+ a->a_nvals[0].bv_val,
+ srs->sr_state.ctxcsn.bv_val );
return LDAP_SUCCESS;
+ }
}
rs->sr_ctrls = op->o_tmpalloc( sizeof(LDAPControl *)*2,
op->o_tmpmemctx );
op->o_tmpfree( cookie.bv_val, op->o_tmpmemctx );
} else {
/* It's RefreshAndPersist, transition to Persist phase */
- syncprov_sendinfo( op, rs, ( ss->ss_present && rs->sr_nentries ) ?
+ syncprov_sendinfo( op, rs, ss->ss_present ?
LDAP_TAG_SYNC_REFRESH_PRESENT : LDAP_TAG_SYNC_REFRESH_DELETE,
&cookie, 1, NULL, 0 );
op->o_tmpfree( cookie.bv_val, op->o_tmpmemctx );
ss->ss_so->s_flags ^= PS_IS_REFRESHING;
syncprov_detach_op( op, ss->ss_so, on );
+
+ /* If there are queued responses, fire them off */
+ if ( ss->ss_so->s_res )
+ syncprov_qstart( ss->ss_so );
ldap_pvt_thread_mutex_unlock( &ss->ss_so->s_mutex );
return LDAP_SUCCESS;
/* syncprov_findbase expects to be called as a callback... */
sc.sc_private = &opc;
opc.son = on;
+ ldap_pvt_thread_mutex_init( &so.s_mutex );
cb = op->o_callback;
op->o_callback = ≻
rs->sr_err = syncprov_findbase( op, &fc );
op->o_callback = cb;
+ ldap_pvt_thread_mutex_destroy( &so.s_mutex );
if ( rs->sr_err != LDAP_SUCCESS ) {
send_ldap_result( op, rs );
sl=si->si_logs;
if ( sl ) {
ldap_pvt_thread_mutex_lock( &sl->sl_mutex );
- if ( ber_bvcmp( &srs->sr_state.ctxcsn, &sl->sl_mincsn ) >= 0 ) {
+ /* Are there any log entries, and is the consumer state
+ * present in the session log?
+ */
+ if ( sl->sl_num > 0 && ber_bvcmp( &srs->sr_state.ctxcsn, &sl->sl_mincsn ) >= 0 ) {
do_present = 0;
/* mutex is unlocked in playlog */
- syncprov_playlog( op, rs, sl, &srs->sr_state.ctxcsn, &ctxcsn );
+ syncprov_playlog( op, rs, sl, srs, &ctxcsn );
} else {
ldap_pvt_thread_mutex_unlock( &sl->sl_mutex );
}
fava->f_next = op->ors_filter;
op->ors_filter = fand;
filter2bv_x( op, op->ors_filter, &op->ors_filterstr );
+ if ( sop )
+ sop->s_flags |= PS_FIX_FILTER;
}
/* Let our callback add needed info to returned entries */
int rc;
void *thrctx = NULL;
+ if ( !SLAP_LASTMOD( be )) {
+ Debug( LDAP_DEBUG_ANY,
+ "syncprov_db_open: invalid config, lastmod must be enabled\n", 0, 0, 0 );
+ return -1;
+ }
+
if ( slapMode & SLAP_TOOL_MODE ) {
return 0;
}
ldap_pvt_thread_create( &tid, 0, syncprov_db_otask, op );
ldap_pvt_thread_join( tid, NULL );
}
- } else if ( SLAP_SYNC_SHADOW( op->o_bd )) {
- /* If we're also a consumer, and we didn't find the context entry,
- * then don't generate anything, wait for our provider to send it
- * to us.
- */
- goto out;
}
if ( BER_BVISEMPTY( &si->si_ctxcsn ) ) {
+ if ( SLAP_SYNC_SHADOW( op->o_bd )) {
+ /* If we're also a consumer, and we didn't get a contextCSN,
+ * then don't generate anything, wait for our provider to send it
+ * to us.
+ */
+ goto out;
+ }
si->si_ctxcsn.bv_len = sizeof( si->si_ctxcsnbuf );
slap_get_csn( op, &si->si_ctxcsn, 0 );
}
if ( si ) {
if ( si->si_logs ) {
+ slog_entry *se = si->si_logs->sl_head;
+
+ while ( se ) {
+ slog_entry *se_next = se->se_next;
+ ch_free( se );
+ se = se_next;
+ }
+
ch_free( si->si_logs );
}
ldap_pvt_thread_mutex_destroy( &si->si_mods_mutex );