syncprov.c

ldap服务器源码

		fop.ors_attrsonly = 0;
		fop.o_managedsait = SLAP_CONTROL_CRITICAL;

		af.f_choice = LDAP_FILTER_AND;
		af.f_next = NULL;
		af.f_and = &mf;
		mf.f_choice = LDAP_FILTER_EQUALITY;
		mf.f_ava = &eq;
		mf.f_av_desc = slap_schema.si_ad_entryUUID;
		mf.f_next = fop.ors_filter;

		fop.ors_filter = ⁡

		cb.sc_response = playlog_cb;
		fop.o_bd->bd_info = on->on_info->oi_orig;

		for ( i=ndel; i<num; i++ ) {
			if ( uuids[i].bv_len == 0 ) continue;

			mf.f_av_value = uuids[i];
			cb.sc_private = NULL;
			fop.ors_slimit = 1;
			frs.sr_nentries = 0;
			rc = fop.o_bd->be_search( &fop, &frs );

			/* If entry was not found, add to delete list */
			if ( !cb.sc_private ) {
				uuids[ndel++] = uuids[i];
			}
		}
		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, &cookie, 0, uuids, 1 );
		op->o_tmpfree( cookie.bv_val, op->o_tmpmemctx );
	}
	op->o_tmpfree( uuids, op->o_tmpmemctx );
}

static int
syncprov_op_response( Operation *op, SlapReply *rs )
{
	opcookie *opc = op->o_callback->sc_private;
	slap_overinst *on = opc->son;
	syncprov_info_t		*si = on->on_bi.bi_private;
	syncmatches *sm;

	if ( rs->sr_err == LDAP_SUCCESS )
	{
		struct berval maxcsn = BER_BVNULL;
		char cbuf[LDAP_LUTIL_CSNSTR_BUFSIZE];
		int do_check=0;

		/* Update our context CSN */
		cbuf[0] = '\0';
		ldap_pvt_thread_rdwr_wlock( &si->si_csn_rwlock );
		slap_get_commit_csn( op, &maxcsn );
		if ( !BER_BVISNULL( &maxcsn ) ) {
			strcpy( cbuf, maxcsn.bv_val );
			if ( ber_bvcmp( &maxcsn, &si->si_ctxcsn ) > 0 ) {
				strcpy( si->si_ctxcsnbuf, cbuf );
				si->si_ctxcsn.bv_len = maxcsn.bv_len;
			}
		}

		/* Don't do any processing for consumer contextCSN updates */
		if ( SLAP_SYNC_SHADOW( op->o_bd ) && 
			op->o_msgid == SLAP_SYNC_UPDATE_MSGID ) {
			ldap_pvt_thread_rdwr_wunlock( &si->si_csn_rwlock );
			return SLAP_CB_CONTINUE;
		}

		si->si_numops++;
		if ( si->si_chkops || si->si_chktime ) {
			if ( si->si_chkops && si->si_numops >= si->si_chkops ) {
				do_check = 1;
				si->si_numops = 0;
			}
			if ( si->si_chktime &&
				(op->o_time - si->si_chklast >= si->si_chktime )) {
				do_check = 1;
				si->si_chklast = op->o_time;
			}
		}
		ldap_pvt_thread_rdwr_wunlock( &si->si_csn_rwlock );

		if ( do_check ) {
			ldap_pvt_thread_rdwr_rlock( &si->si_csn_rwlock );
			syncprov_checkpoint( op, rs, on );
			ldap_pvt_thread_rdwr_runlock( &si->si_csn_rwlock );
		}

		opc->sctxcsn.bv_len = maxcsn.bv_len;
		opc->sctxcsn.bv_val = cbuf;

		/* Handle any persistent searches */
		if ( si->si_ops ) {
			switch(op->o_tag) {
			case LDAP_REQ_ADD:
			case LDAP_REQ_MODIFY:
			case LDAP_REQ_MODRDN:
			case LDAP_REQ_EXTENDED:
				syncprov_matchops( op, opc, 0 );
				break;
			case LDAP_REQ_DELETE:
				/* for each match in opc->smatches:
				 *   send DELETE msg
				 */
				for ( sm = opc->smatches; sm; sm=sm->sm_next ) {
					if ( sm->sm_op->s_op->o_abandon )
						continue;
					syncprov_qresp( opc, sm->sm_op, LDAP_SYNC_DELETE );
				}
				break;
			}
		}

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

	}
	return SLAP_CB_CONTINUE;
}

/* We don't use a subentry to store the context CSN any more.
 * We expose the current context CSN as an operational attribute
 * of the suffix entry.
 */
static int
syncprov_op_compare( Operation *op, SlapReply *rs )
{
	slap_overinst		*on = (slap_overinst *)op->o_bd->bd_info;
	syncprov_info_t		*si = on->on_bi.bi_private;
	int rc = SLAP_CB_CONTINUE;

	if ( dn_match( &op->o_req_ndn, op->o_bd->be_nsuffix ) &&
		op->oq_compare.rs_ava->aa_desc == slap_schema.si_ad_contextCSN )
	{
		Entry e = {0};
		Attribute a = {0};
		struct berval bv[2];

		e.e_name = op->o_bd->be_suffix[0];
		e.e_nname = op->o_bd->be_nsuffix[0];

		BER_BVZERO( &bv[1] );
		bv[0] = si->si_ctxcsn;

		a.a_desc = slap_schema.si_ad_contextCSN;
		a.a_vals = bv;
		a.a_nvals = a.a_vals;

		ldap_pvt_thread_rdwr_rlock( &si->si_csn_rwlock );

		rs->sr_err = access_allowed( op, &e, op->oq_compare.rs_ava->aa_desc,
			&op->oq_compare.rs_ava->aa_value, ACL_COMPARE, NULL );
		if ( ! rs->sr_err ) {
			rs->sr_err = LDAP_INSUFFICIENT_ACCESS;
			goto return_results;
		}

		if ( get_assert( op ) &&
			( test_filter( op, &e, get_assertion( op ) ) != LDAP_COMPARE_TRUE ) )
		{
			rs->sr_err = LDAP_ASSERTION_FAILED;
			goto return_results;
		}


		rs->sr_err = LDAP_COMPARE_FALSE;

		if ( value_find_ex( op->oq_compare.rs_ava->aa_desc,
			SLAP_MR_ATTRIBUTE_VALUE_NORMALIZED_MATCH |
				SLAP_MR_ASSERTED_VALUE_NORMALIZED_MATCH,
				a.a_nvals, &op->oq_compare.rs_ava->aa_value, op->o_tmpmemctx ) == 0 )
		{
			rs->sr_err = LDAP_COMPARE_TRUE;
		}

return_results:;

		ldap_pvt_thread_rdwr_runlock( &si->si_csn_rwlock );

		send_ldap_result( op, rs );

		if( rs->sr_err == LDAP_COMPARE_FALSE || rs->sr_err == LDAP_COMPARE_TRUE ) {
			rs->sr_err = LDAP_SUCCESS;
		}
		rc = rs->sr_err;
	}

	return rc;
}

static int
syncprov_op_mod( Operation *op, SlapReply *rs )
{
	slap_overinst		*on = (slap_overinst *)op->o_bd->bd_info;
	syncprov_info_t		*si = on->on_bi.bi_private;

	slap_callback *cb = op->o_tmpcalloc(1, sizeof(slap_callback)+
		sizeof(opcookie) +
		(si->si_ops ? sizeof(modinst) : 0 ),
		op->o_tmpmemctx);
	opcookie *opc = (opcookie *)(cb+1);
	opc->son = on;
	cb->sc_response = syncprov_op_response;
	cb->sc_cleanup = syncprov_op_cleanup;
	cb->sc_private = opc;
	cb->sc_next = op->o_callback;
	op->o_callback = cb;

	/* If there are active persistent searches, lock this operation.
	 * See seqmod.c for the locking logic on its own.
	 */
	if ( si->si_ops ) {
		modtarget *mt, mtdummy;
		modinst *mi;

		mi = (modinst *)(opc+1);
		mi->mi_op = op;

		/* See if we're already modifying this entry... */
		mtdummy.mt_op = op;
		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 );
			ldap_pvt_thread_mutex_unlock( &si->si_mods_mutex );
			mt->mt_tail->mi_next = mi;
			mt->mt_tail = mi;
			/* wait for this op to get to head of list */
			while ( mt->mt_mods != mi ) {
				ldap_pvt_thread_mutex_unlock( &mt->mt_mutex );
				ldap_pvt_thread_yield();
				ldap_pvt_thread_mutex_lock( &mt->mt_mutex );

				/* 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;
					op->o_tmpfree( cb, op->o_tmpmemctx );
					ldap_pvt_thread_mutex_unlock( &mt->mt_mutex );
					return SLAPD_ABANDON;
				}
			}
			ldap_pvt_thread_mutex_unlock( &mt->mt_mutex );
		} else {
			/* Record that we're modifying this entry now */
			mt = ch_malloc( sizeof(modtarget) );
			mt->mt_mods = mi;
			mt->mt_tail = mi;
			mt->mt_op = mi->mi_op;
			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 );
		}
	}

	if (( si->si_ops || si->si_logs ) && op->o_tag != LDAP_REQ_ADD )
		syncprov_matchops( op, opc, 1 );

	return SLAP_CB_CONTINUE;
}

static int
syncprov_op_extended( Operation *op, SlapReply *rs )
{
	if ( exop_is_write( op ))
		return syncprov_op_mod( op, rs );

	return SLAP_CB_CONTINUE;
}

typedef struct searchstate {
	slap_overinst *ss_on;
	syncops *ss_so;
	int ss_present;
	struct berval ss_ctxcsn;
	char ss_csnbuf[LDAP_LUTIL_CSNSTR_BUFSIZE];
} 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;
}

static void
syncprov_detach_op( Operation *op, syncops *so, slap_overinst *on )
{
	Operation *op2;
	int i, alen = 0;
	size_t size;
	char *ptr;
	GroupAssertion *g1, *g2;

	/* count the search attrs */
	for (i=0; op->ors_attrs && !BER_BVISNULL( &op->ors_attrs[i].an_name ); i++) {
		alen += op->ors_attrs[i].an_name.bv_len + 1;
	}
	/* Make a new copy of the operation */
	size = sizeof(Operation) + sizeof(Opheader) +
		(i ? ( (i+1) * sizeof(AttributeName) + alen) : 0) +
		op->o_req_dn.bv_len + 1 +
		op->o_req_ndn.bv_len + 1 +
		op->o_ndn.bv_len + 1 +
		so->s_filterstr.bv_len + 1;
	op2 = (Operation *)ch_calloc( 1, size );
	op2->o_hdr = (Opheader *)(op2+1);

	/* Copy the fields we care about explicitly, leave the rest alone */
	*op2->o_hdr = *op->o_hdr;
	op2->o_tag = op->o_tag;
	op2->o_time = op->o_time;
	op2->o_bd = on->on_info->oi_origdb;
	op2->o_request = op->o_request;
	op2->o_private = on;

	if ( i ) {
		op2->ors_attrs = (AttributeName *)(op2->o_hdr + 1);
		ptr = (char *)(op2->ors_attrs+i+1);
		for (i=0; !BER_BVISNULL( &op->ors_attrs[i].an_name ); i++) {
			op2->ors_attrs[i] = op->ors_attrs[i];
			op2->ors_attrs[i].an_name.bv_val = ptr;
			ptr = lutil_strcopy( ptr, op->ors_attrs[i].an_name.bv_val ) + 1;
		}
		BER_BVZERO( &op2->ors_attrs[i].an_name );
	} else {
		ptr = (char *)(op2->o_hdr + 1);
	}
	op2->o_authz = op->o_authz;
	op2->o_ndn.bv_val = ptr;
	ptr = lutil_strcopy(ptr, op->o_ndn.bv_val) + 1;
	op2->o_dn = op2->o_ndn;
	op2->o_req_dn.bv_len = op->o_req_dn.bv_len;
	op2->o_req_dn.bv_val = ptr;
	ptr = lutil_strcopy(ptr, op->o_req_dn.bv_val) + 1;
	op2->o_req_ndn.bv_len = op->o_req_ndn.bv_len;
	op2->o_req_ndn.bv_val = ptr;
	ptr = lutil_strcopy(ptr, op->o_req_ndn.bv_val) + 1;
	op2->ors_filterstr.bv_val = ptr;
	strcpy( ptr, so->s_filterstr.bv_val );
	op2->ors_filterstr.bv_len = so->s_filterstr.bv_len;

	/* 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 */
	op2->o_groups = NULL;
	for ( g1=op->o_groups; g1; g1=g1->ga_next ) {
		g2 = ch_malloc( sizeof(GroupAssertion) + g1->ga_len );
		*g2 = *g1;
		strcpy( g2->ga_ndn, g1->ga_ndn );
		g2->ga_next = op2->o_groups;
		op2->o_groups = g2;
	}
	/* Don't allow any further group caching */
	op2->o_do_not_cache = 1;

	/* Add op2 to conn so abandon will find us */
	ldap_pvt_thread_mutex_lock( &op->o_conn->c_mutex );
	op->o_conn->c_n_ops_executing++;
	op->o_conn->c_n_ops_completed--;
	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
syncprov_search_response( Operation *op, SlapReply *rs )
{
	searchstate *ss = op->o_callback->sc_private;
	slap_overinst *on = ss->ss_on;
	sync_control *srs = op->o_controls[slap_cids.sc_LDAPsync];

	if ( rs->sr_type == REP_SEARCH || rs->sr_type == REP_SEARCHREF ) {
		Attribute *a;
		/* If we got a referral without a referral object, there's
		 * something missing that we cannot replicate. Just ignore it.
		 * The consumer will abort because we didn't send the expected
		 * control.
		 */
		if ( !rs->sr_entry ) {
			assert( rs->sr_entry != NULL );
			Debug( LDAP_DEBUG_ANY, "bogus referral in context\n",0,0,0 );
			return SLAP_CB_CONTINUE;
		}
		a = attr_find( rs->sr_entry->e_attrs, slap_schema.si_ad_entryCSN );
		if ( a == NULL && rs->sr_operational_attrs != NULL ) {
			a = attr_find( rs->sr_operational_attrs, slap_schema.si_ad_entryCSN );
		}
		if ( a ) {
			/* 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 ) ) {
				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 );
		rs->sr_ctrls[1] = NULL;
		rs->sr_err = syncprov_state_ctrl( op, rs, rs->sr_entry,
			LDAP_SYNC_ADD, rs->sr_ctrls, 0, 0, NULL );
	} else if ( rs->sr_type == REP_RESULT && rs->sr_err == LDAP_SUCCESS ) {
		struct berval cookie;

		slap_compose_sync_cookie( op, &cookie, &ss->ss_ctxcsn,
			srs->sr_state.rid );

		/* Is this a regular refresh? */
		if ( !ss->ss_so ) {
			rs->sr_ctrls = op->o_tmpalloc( sizeof(LDAPControl *)*2,
				op->o_tmpmemctx );
			rs->sr_ctrls[1] = NULL;
			rs->sr_err = syncprov_done_ctrl( op, rs, rs->sr_ctrls,
				0, 1, &cookie, ss->ss_present ?  LDAP_SYNC_REFRESH_PRESENTS :
					LDAP_SYNC_REFRESH_DELETES );
			op->o_tmpfree( cookie.bv_val, op->o_tmpmemctx );
		} else {
		/* It's RefreshAndPersist, transition to Persist phase */
			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 );

			/* Detach this Op from frontend control */
			ldap_pvt_thread_mutex_lock( &ss->ss_so->s_mutex );

			/* Turn off the refreshing flag */
			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;
		}
	}

	return SLAP_CB_CONTINUE;
}

static int