servers/slapd/overlays/syncprov.c - diff

Return to syncprov.c CVS log

Up to [OpenLDAP] / servers / slapd / overlays

Diff for /servers/slapd/overlays/syncprov.c between versions 1.147.2.1 and 1.147.2.2

version 1.147.2.1, 2006/05/15 17:04:44	version 1.147.2.2, 2006/08/17 23:50:32
Line 1	Line 1
/* $OpenLDAP: pkg/ldap/servers/slapd/overlays/syncprov.c,v 1.150 2006/05/11 08:46:44 hyc Exp $ */	/* $OpenLDAP: pkg/ldap/servers/slapd/overlays/syncprov.c,v 1.147.2.1 2006/05/15 17:04:44 kurt Exp $ */
/* syncprov.c - syncrepl provider */	/* syncprov.c - syncrepl provider */
/* This work is part of OpenLDAP Software <http://www.openldap.org/>.	/* This work is part of OpenLDAP Software <http://www.openldap.org/>.
*	*
Line 65 typedef struct syncops {	Line 65 typedef struct syncops {
#define PS_IS_DETACHED 0x02	#define PS_IS_DETACHED 0x02
#define PS_WROTE_BASE 0x04	#define PS_WROTE_BASE 0x04
#define PS_FIND_BASE 0x08	#define PS_FIND_BASE 0x08
	#define PS_FIX_FILTER 0x10

int s_inuse; /* reference count */	int s_inuse; /* reference count */
struct syncres *s_res;	struct syncres *s_res;
Line 911 syncprov_qtask( void ctx, void arg )	Line 912 syncprov_qtask( void ctx, void arg )
return NULL;	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 */	/* 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 )
Line 949 syncprov_qresp( opcookie opc, syncops	Line 977 syncprov_qresp( opcookie opc, syncops
so->s_flags \|= PS_FIND_BASE;	so->s_flags \|= PS_FIND_BASE;
}	}
if ( so->s_flags & PS_IS_DETACHED ) {	if ( so->s_flags & PS_IS_DETACHED ) {
int wake=0;	syncprov_qstart( so );
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();
}	}
ldap_pvt_thread_mutex_unlock( &so->s_mutex );	ldap_pvt_thread_mutex_unlock( &so->s_mutex );
return LDAP_SUCCESS;	return LDAP_SUCCESS;
Line 1055 syncprov_matchops( Operation *op, opcook	Line 1063 syncprov_matchops( Operation *op, opcook
int rc;	int rc;
struct berval newdn;	struct berval newdn;
int freefdn = 0;	int freefdn = 0;
	BackendDB *b0 = op->o_bd, db;

fc.fdn = &op->o_req_ndn;	fc.fdn = &op->o_req_ndn;
/* compute new DN */	/* compute new DN */
Line 1067 syncprov_matchops( Operation *op, opcook	Line 1076 syncprov_matchops( Operation *op, opcook
freefdn = 1;	freefdn = 1;
}	}
if ( op->o_tag != LDAP_REQ_ADD ) {	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;	op->o_bd->bd_info = (BackendInfo *)on->on_info;
rc = be_entry_get_rw( op, fc.fdn, NULL, NULL, 0, &e );	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 */	/* If we're sending responses now, make a copy and unlock the DB */
Line 1076 syncprov_matchops( Operation *op, opcook	Line 1089 syncprov_matchops( Operation *op, opcook
e = e2;	e = e2;
}	}
op->o_bd->bd_info = (BackendInfo *)on;	op->o_bd->bd_info = (BackendInfo *)on;
if ( rc ) return;	if ( rc ) {
	op->o_bd = b0;
	return;
	}
} else {	} else {
e = op->ora_e;	e = op->ora_e;
}	}
Line 1174 syncprov_matchops( Operation *op, opcook	Line 1190 syncprov_matchops( Operation *op, opcook
if ( freefdn ) {	if ( freefdn ) {
op->o_tmpfree( fc.fdn->bv_val, op->o_tmpmemctx );	op->o_tmpfree( fc.fdn->bv_val, op->o_tmpmemctx );
}	}
	op->o_bd = b0;
}	}

static int	static int
Line 1235 syncprov_checkpoint( Operation *op, Slap	Line 1252 syncprov_checkpoint( Operation *op, Slap
struct berval bv[2];	struct berval bv[2];
slap_callback cb = {0};	slap_callback cb = {0};

mod.sml_values = bv;	/* If ctxcsn is empty, delete it */
bv[1].bv_val = NULL;	if ( BER_BVISEMPTY( &si->si_ctxcsn )) {
bv[0] = 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_nvalues = NULL;
mod.sml_desc = slap_schema.si_ad_contextCSN;	mod.sml_desc = slap_schema.si_ad_contextCSN;
mod.sml_op = LDAP_MOD_REPLACE;	mod.sml_op = LDAP_MOD_REPLACE;
Line 1262 syncprov_checkpoint( Operation *op, Slap	Line 1284 syncprov_checkpoint( Operation *op, Slap
}	}

static void	static void
syncprov_add_slog( Operation op, struct berval csn )	syncprov_add_slog( Operation *op )
{	{
opcookie *opc = op->o_callback->sc_private;	opcookie *opc = op->o_callback->sc_private;
slap_overinst *on = opc->son;	slap_overinst *on = opc->son;
Line 1274 syncprov_add_slog( Operation *op, struct	Line 1296 syncprov_add_slog( Operation *op, struct
{	{
/* Allocate a record. UUIDs are not NUL-terminated. */	/* Allocate a record. UUIDs are not NUL-terminated. */
se = ch_malloc( sizeof( slog_entry ) + opc->suuid.bv_len +	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_next = NULL;
se->se_tag = op->o_tag;	se->se_tag = op->o_tag;

Line 1283 syncprov_add_slog( Operation *op, struct	Line 1305 syncprov_add_slog( Operation *op, struct
se->se_uuid.bv_len = 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;	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 );	AC_MEMCPY( se->se_csn.bv_val, op->o_csn.bv_val, op->o_csn.bv_len );
se->se_csn.bv_val[csn->bv_len] = '\0';	se->se_csn.bv_val[op->o_csn.bv_len] = '\0';
se->se_csn.bv_len = csn->bv_len;	se->se_csn.bv_len = op->o_csn.bv_len;

ldap_pvt_thread_mutex_lock( &sl->sl_mutex );	ldap_pvt_thread_mutex_lock( &sl->sl_mutex );
if ( sl->sl_head ) {	if ( sl->sl_head ) {
Line 1357 syncprov_playlog( Operation *op, SlapRep	Line 1379 syncprov_playlog( Operation *op, SlapRep
i++;	i++;
AC_MEMCPY( cbuf, se->se_csn.bv_val, se->se_csn.bv_len );	AC_MEMCPY( cbuf, se->se_csn.bv_val, se->se_csn.bv_len );
delcsn.bv_len = se->se_csn.bv_len;	delcsn.bv_len = se->se_csn.bv_len;
	delcsn.bv_val[delcsn.bv_len] = '\0';
} else {	} else {
nmods++;	nmods++;
j = num - nmods;	j = num - nmods;
Line 1468 syncprov_op_response( Operation *op, Sla	Line 1491 syncprov_op_response( Operation *op, Sla

if ( rs->sr_err == LDAP_SUCCESS )	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];	char cbuf[LDAP_LUTIL_CSNSTR_BUFSIZE];

/* Update our context CSN */	/* Update our context CSN */
cbuf[0] = '\0';	cbuf[0] = '\0';
ldap_pvt_thread_mutex_lock( &si->si_csn_mutex );	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 ) ) {	if ( !BER_BVISNULL( &maxcsn ) ) {
strcpy( cbuf, maxcsn.bv_val );	strcpy( cbuf, maxcsn.bv_val );
if ( ber_bvcmp( &maxcsn, &si->si_ctxcsn ) > 0 ) {	if ( ber_bvcmp( &maxcsn, &si->si_ctxcsn ) > 0 ) {
Line 1535 syncprov_op_response( Operation *op, Sla	Line 1558 syncprov_op_response( Operation *op, Sla

/* Add any log records */	/* Add any log records */
if ( si->si_logs && op->o_tag != LDAP_REQ_ADD ) {	if ( si->si_logs && op->o_tag != LDAP_REQ_ADD ) {
syncprov_add_slog( op, &curcsn );	syncprov_add_slog( op );
}	}

}	}
Line 1770 syncprov_detach_op( Operation *op, synco	Line 1793 syncprov_detach_op( Operation *op, synco
op2->ors_filterstr.bv_val = ptr;	op2->ors_filterstr.bv_val = ptr;
strcpy( ptr, so->s_filterstr.bv_val );	strcpy( ptr, so->s_filterstr.bv_val );
op2->ors_filterstr.bv_len = so->s_filterstr.bv_len;	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;	so->s_op = op2;

/* Copy any cached group ACLs individually */	/* Copy any cached group ACLs individually */
Line 1862 syncprov_search_response( Operation *op,	Line 1893 syncprov_search_response( Operation *op,
op->o_tmpfree( cookie.bv_val, op->o_tmpmemctx );	op->o_tmpfree( cookie.bv_val, op->o_tmpmemctx );
} else {	} else {
/* It's RefreshAndPersist, transition to Persist phase */	/* 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,	LDAP_TAG_SYNC_REFRESH_PRESENT : LDAP_TAG_SYNC_REFRESH_DELETE,
&cookie, 1, NULL, 0 );	&cookie, 1, NULL, 0 );
op->o_tmpfree( cookie.bv_val, op->o_tmpmemctx );	op->o_tmpfree( cookie.bv_val, op->o_tmpmemctx );
Line 1874 syncprov_search_response( Operation *op,	Line 1905 syncprov_search_response( Operation *op,
ss->ss_so->s_flags ^= PS_IS_REFRESHING;	ss->ss_so->s_flags ^= PS_IS_REFRESHING;

syncprov_detach_op( op, ss->ss_so, on );	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 );	ldap_pvt_thread_mutex_unlock( &ss->ss_so->s_mutex );

return LDAP_SUCCESS;	return LDAP_SUCCESS;
Line 1981 syncprov_op_search( Operation *op, SlapR	Line 2016 syncprov_op_search( Operation *op, SlapR
sl=si->si_logs;	sl=si->si_logs;
if ( sl ) {	if ( sl ) {
ldap_pvt_thread_mutex_lock( &sl->sl_mutex );	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;	do_present = 0;
/* mutex is unlocked in playlog */	/* mutex is unlocked in playlog */
syncprov_playlog( op, rs, sl, srs, &ctxcsn );	syncprov_playlog( op, rs, sl, srs, &ctxcsn );
Line 2035 shortcut:	Line 2073 shortcut:
fava->f_next = op->ors_filter;	fava->f_next = op->ors_filter;
op->ors_filter = fand;	op->ors_filter = fand;
filter2bv_x( op, op->ors_filter, &op->ors_filterstr );	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 */	/* Let our callback add needed info to returned entries */
Line 2317 syncprov_db_open(	Line 2357 syncprov_db_open(
int rc;	int rc;
void *thrctx = NULL;	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 ) {	if ( slapMode & SLAP_TOOL_MODE ) {
return 0;	return 0;
}	}
Line 2360 syncprov_db_open(	Line 2406 syncprov_db_open(
ldap_pvt_thread_create( &tid, 0, syncprov_db_otask, op );	ldap_pvt_thread_create( &tid, 0, syncprov_db_otask, op );
ldap_pvt_thread_join( tid, NULL );	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 ( 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 );	si->si_ctxcsn.bv_len = sizeof( si->si_ctxcsnbuf );
slap_get_csn( op, &si->si_ctxcsn, 0 );	slap_get_csn( op, &si->si_ctxcsn, 0 );
}	}

Removed from v.1.147.2.1
changed lines
	Added in v.1.147.2.2