/* syncprov.c - syncrepl provider */
/* This work is part of OpenLDAP Software <http://www.openldap.org/>.
*
- * Copyright 2004-2006 The OpenLDAP Foundation.
+ * Copyright 2004-2007 The OpenLDAP Foundation.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
#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;
rs.sr_flags = REP_ENTRY_MUSTRELEASE;
if ( opc->sreference ) {
rs.sr_ref = get_entry_referrals( op, rs.sr_entry );
- send_search_reference( op, &rs );
+ rs.sr_err = send_search_reference( op, &rs );
ber_bvarray_free( rs.sr_ref );
if ( !rs.sr_entry )
*e = NULL;
if ( rs.sr_entry->e_private )
rs.sr_flags = REP_ENTRY_MUSTRELEASE;
rs.sr_attrs = op->ors_attrs;
- send_search_entry( op, &rs );
+ rs.sr_err = send_search_entry( op, &rs );
if ( !rs.sr_entry )
*e = NULL;
break;
if ( opc->sreference ) {
struct berval bv = BER_BVNULL;
rs.sr_ref = &bv;
- send_search_reference( op, &rs );
+ rs.sr_err = send_search_reference( op, &rs );
} else {
- send_search_entry( op, &rs );
+ rs.sr_err = send_search_entry( op, &rs );
}
break;
default:
OperationBuffer opbuf;
Operation *op;
BackendDB be;
+ int rc;
op = (Operation *) &opbuf;
*op = *so->s_op;
op->o_private = NULL;
op->o_callback = NULL;
- (void)syncprov_qplay( op, on, so );
+ rc = syncprov_qplay( op, on, so );
/* decrement use count... */
syncprov_free_syncop( so );
/* wait until we get explicitly scheduled again */
ldap_pvt_thread_mutex_lock( &slapd_rq.rq_mutex );
- ldap_pvt_runqueue_stoptask( &slapd_rq, so->s_qtask );
- ldap_pvt_runqueue_resched( &slapd_rq, so->s_qtask, 1 );
+ ldap_pvt_runqueue_stoptask( &slapd_rq, rtask );
+ if ( rc == 0 ) {
+ ldap_pvt_runqueue_resched( &slapd_rq, rtask, 1 );
+ } else {
+ /* bail out on any error */
+ ldap_pvt_runqueue_remove( &slapd_rq, rtask );
+ }
ldap_pvt_thread_mutex_unlock( &slapd_rq.rq_mutex );
+#if 0 /* FIXME: connection_close isn't exported from slapd.
+ * should it be?
+ */
+ if ( rc ) {
+ ldap_pvt_thread_mutex_lock( &op->o_conn->c_mutex );
+ if ( connection_state_closing( op->o_conn )) {
+ connection_close( op->o_conn );
+ }
+ ldap_pvt_thread_mutex_unlock( &op->o_conn->c_mutex );
+ }
+#endif
return NULL;
}
}
static void
-syncprov_checkpoint( Operation *op, SlapReply *rs, slap_overinst *on )
+syncprov_checkpoint( Operation *op, SlapReply *rs, slap_overinst *on,
+ struct berval *csn )
{
- syncprov_info_t *si = on->on_bi.bi_private;
Modifications mod;
Operation opm;
SlapReply rsm = { 0 };
slap_callback cb = {0};
/* If ctxcsn is empty, delete it */
- if ( BER_BVISEMPTY( &si->si_ctxcsn )) {
+ if ( BER_BVISEMPTY( csn )) {
mod.sml_values = NULL;
} else {
mod.sml_values = bv;
bv[1].bv_val = NULL;
- bv[0] = si->si_ctxcsn;
+ bv[0] = *csn;
}
mod.sml_nvalues = NULL;
mod.sml_desc = slap_schema.si_ad_contextCSN;
{
struct berval maxcsn = BER_BVNULL;
char cbuf[LDAP_LUTIL_CSNSTR_BUFSIZE];
+ int do_check = 0;
/* Update our context CSN */
cbuf[0] = '\0';
si->si_numops++;
if ( si->si_chkops || si->si_chktime ) {
- int do_check=0;
if ( si->si_chkops && si->si_numops >= si->si_chkops ) {
do_check = 1;
si->si_numops = 0;
do_check = 1;
si->si_chklast = op->o_time;
}
- if ( do_check ) {
- syncprov_checkpoint( op, rs, on );
- }
}
ldap_pvt_thread_mutex_unlock( &si->si_csn_mutex );
opc->sctxcsn.bv_len = maxcsn.bv_len;
opc->sctxcsn.bv_val = cbuf;
+ if ( do_check ) {
+ syncprov_checkpoint( op, rs, on, &opc->sctxcsn );
+ }
+
/* Handle any persistent searches */
if ( si->si_ops ) {
switch(op->o_tag) {
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 */
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 );
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 */
if ( !a ) {
for ( ap = &rs->sr_operational_attrs; *ap; ap=&(*ap)->a_next );
- a = ch_malloc( sizeof(Attribute));
- a->a_desc = slap_schema.si_ad_contextCSN;
+ a = attr_alloc( slap_schema.si_ad_contextCSN );
a->a_vals = ch_malloc( 2 * sizeof(struct berval));
a->a_vals[1].bv_val = NULL;
a->a_nvals = a->a_vals;
- a->a_next = NULL;
- a->a_flags = 0;
*ap = a;
}
switch ( c->type ) {
case SP_CHKPT:
if ( lutil_atoi( &si->si_chkops, c->argv[1] ) != 0 ) {
- sprintf( c->msg, "%s unable to parse checkpoint ops # \"%s\"",
+ snprintf( c->msg, sizeof( c->msg ), "%s unable to parse checkpoint ops # \"%s\"",
c->argv[0], c->argv[1] );
- Debug( LDAP_DEBUG_CONFIG, "%s: %s\n", c->log, c->msg, 0 );
+ Debug( LDAP_DEBUG_CONFIG|LDAP_DEBUG_NONE,
+ "%s: %s\n", c->log, c->msg, 0 );
return ARG_BAD_CONF;
}
if ( si->si_chkops <= 0 ) {
- sprintf( c->msg, "%s invalid checkpoint ops # \"%d\"",
+ snprintf( c->msg, sizeof( c->msg ), "%s invalid checkpoint ops # \"%d\"",
c->argv[0], si->si_chkops );
- Debug( LDAP_DEBUG_CONFIG, "%s: %s\n", c->log, c->msg, 0 );
+ Debug( LDAP_DEBUG_CONFIG|LDAP_DEBUG_NONE,
+ "%s: %s\n", c->log, c->msg, 0 );
return ARG_BAD_CONF;
}
if ( lutil_atoi( &si->si_chktime, c->argv[2] ) != 0 ) {
- sprintf( c->msg, "%s unable to parse checkpoint time \"%s\"",
+ snprintf( c->msg, sizeof( c->msg ), "%s unable to parse checkpoint time \"%s\"",
c->argv[0], c->argv[1] );
- Debug( LDAP_DEBUG_CONFIG, "%s: %s\n", c->log, c->msg, 0 );
+ Debug( LDAP_DEBUG_CONFIG|LDAP_DEBUG_NONE,
+ "%s: %s\n", c->log, c->msg, 0 );
return ARG_BAD_CONF;
}
if ( si->si_chktime <= 0 ) {
- sprintf( c->msg, "%s invalid checkpoint time \"%d\"",
+ snprintf( c->msg, sizeof( c->msg ), "%s invalid checkpoint time \"%d\"",
c->argv[0], si->si_chkops );
- Debug( LDAP_DEBUG_CONFIG, "%s: %s\n", c->log, c->msg, 0 );
+ Debug( LDAP_DEBUG_CONFIG|LDAP_DEBUG_NONE,
+ "%s: %s\n", c->log, c->msg, 0 );
return ARG_BAD_CONF;
}
si->si_chktime *= 60;
int size = c->value_int;
if ( size < 0 ) {
- sprintf( c->msg, "%s size %d is negative",
+ snprintf( c->msg, sizeof( c->msg ), "%s size %d is negative",
c->argv[0], size );
- Debug( LDAP_DEBUG_CONFIG, "%s: %s\n", c->log, c->msg, 0 );
+ Debug( LDAP_DEBUG_CONFIG|LDAP_DEBUG_NONE,
+ "%s: %s\n", c->log, c->msg, 0 );
return ARG_BAD_CONF;
}
sl = si->si_logs;
op->o_bd = be;
op->o_dn = be->be_rootdn;
op->o_ndn = be->be_rootndn;
- syncprov_checkpoint( op, &rs, on );
+ syncprov_checkpoint( op, &rs, on, &si->si_ctxcsn );
ldap_pvt_thread_pool_context_reset( thrctx );
}
slap_overinst *on = (slap_overinst *)be->bd_info;
syncprov_info_t *si;
+ if ( SLAP_ISGLOBALOVERLAY( be ) ) {
+ Debug( LDAP_DEBUG_ANY,
+ "syncprov must be instantiated within a database.\n",
+ 0, 0, 0 );
+ return 1;
+ }
+
si = ch_calloc(1, sizeof(syncprov_info_t));
on->on_bi.bi_private = si;
ldap_pvt_thread_mutex_init( &si->si_csn_mutex );
projects / openldap / blobdiff