syncprov.c

ldap服务器源码

	if ( !so->s_res ) {
		so->s_res = sr;
	} else {
		so->s_restail->s_next = sr;
	}
	so->s_restail = sr;

	/* If the base of the psearch was modified, check it next time round */
	if ( so->s_flags & PS_WROTE_BASE ) {
		so->s_flags ^= PS_WROTE_BASE;
		so->s_flags |= PS_FIND_BASE;
	}
	if ( so->s_flags & PS_IS_DETACHED ) {
		syncprov_qstart( so );
	}
	ldap_pvt_thread_mutex_unlock( &so->s_mutex );
	return LDAP_SUCCESS;
}

static int
syncprov_drop_psearch( syncops *so, int lock )
{
	if ( so->s_flags & PS_IS_DETACHED ) {
		if ( lock )
			ldap_pvt_thread_mutex_lock( &so->s_op->o_conn->c_mutex );
		so->s_op->o_conn->c_n_ops_executing--;
		so->s_op->o_conn->c_n_ops_completed++;
		LDAP_STAILQ_REMOVE( &so->s_op->o_conn->c_ops, so->s_op, slap_op,
			o_next );
		if ( lock )
			ldap_pvt_thread_mutex_unlock( &so->s_op->o_conn->c_mutex );
	}
	syncprov_free_syncop( so );

	return 0;
}

static int
syncprov_ab_cleanup( Operation *op, SlapReply *rs )
{
	slap_callback *sc = op->o_callback;
	op->o_callback = sc->sc_next;
	syncprov_drop_psearch( op->o_callback->sc_private, 0 );
	op->o_tmpfree( sc, op->o_tmpmemctx );
	return 0;
}

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

	ldap_pvt_thread_mutex_lock( &si->si_ops_mutex );
	for ( so=si->si_ops, soprev = (syncops *)&si->si_ops; so;
		soprev=so, so=so->s_next ) {
		if ( so->s_op->o_connid == op->o_connid &&
			so->s_op->o_msgid == op->orn_msgid ) {
				so->s_op->o_abandon = 1;
				soprev->s_next = so->s_next;
				break;
		}
	}
	ldap_pvt_thread_mutex_unlock( &si->si_ops_mutex );
	if ( so ) {
		/* Is this really a Cancel exop? */
		if ( op->o_tag != LDAP_REQ_ABANDON ) {
			so->s_op->o_cancel = SLAP_CANCEL_ACK;
			rs->sr_err = LDAP_CANCELLED;
			send_ldap_result( so->s_op, rs );
			if ( so->s_flags & PS_IS_DETACHED ) {
				slap_callback *cb;
				cb = op->o_tmpcalloc( 1, sizeof(slap_callback), op->o_tmpmemctx );
				cb->sc_cleanup = syncprov_ab_cleanup;
				cb->sc_next = op->o_callback;
				cb->sc_private = so;
				return SLAP_CB_CONTINUE;
			}
		}
		syncprov_drop_psearch( so, 0 );
	}
	return SLAP_CB_CONTINUE;
}

/* Find which persistent searches are affected by this operation */
static void
syncprov_matchops( Operation *op, opcookie *opc, int saveit )
{
	slap_overinst *on = opc->son;
	syncprov_info_t		*si = on->on_bi.bi_private;

	fbase_cookie fc;
	syncops *ss, *sprev, *snext;
	Entry *e;
	Attribute *a;
	int rc;
	struct berval newdn;
	int freefdn = 0;
	BackendDB *b0 = op->o_bd, db;

	fc.fdn = &op->o_req_ndn;
	/* compute new DN */
	if ( op->o_tag == LDAP_REQ_MODRDN && !saveit ) {
		struct berval pdn;
		if ( op->orr_nnewSup ) pdn = *op->orr_nnewSup;
		else dnParent( fc.fdn, &pdn );
		build_new_dn( &newdn, &pdn, &op->orr_nnewrdn, op->o_tmpmemctx );
		fc.fdn = &newdn;
		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 */
		if ( e && !saveit ) {
			Entry *e2 = entry_dup( e );
			be_entry_release_rw( op, e, 0 );
			e = e2;
		}
		op->o_bd->bd_info = (BackendInfo *)on;
		if ( rc ) {
			op->o_bd = b0;
			return;
		}
	} else {
		e = op->ora_e;
	}

	if ( saveit || op->o_tag == LDAP_REQ_ADD ) {
		ber_dupbv_x( &opc->sdn, &e->e_name, op->o_tmpmemctx );
		ber_dupbv_x( &opc->sndn, &e->e_nname, op->o_tmpmemctx );
		opc->sreference = is_entry_referral( e );
		a = attr_find( e->e_attrs, slap_schema.si_ad_entryUUID );
		if ( a )
			ber_dupbv_x( &opc->suuid, &a->a_nvals[0], op->o_tmpmemctx );
	} else if ( op->o_tag == LDAP_REQ_MODRDN && !saveit ) {
		op->o_tmpfree( opc->sndn.bv_val, op->o_tmpmemctx );
		op->o_tmpfree( opc->sdn.bv_val, op->o_tmpmemctx );
		ber_dupbv_x( &opc->sdn, &e->e_name, op->o_tmpmemctx );
		ber_dupbv_x( &opc->sndn, &e->e_nname, op->o_tmpmemctx );
	}

	ldap_pvt_thread_mutex_lock( &si->si_ops_mutex );
	for (ss = si->si_ops, sprev = (syncops *)&si->si_ops; ss;
		sprev = ss, ss=snext)
	{
		syncmatches *sm;
		int found = 0;

		snext = ss->s_next;
		/* validate base */
		fc.fss = ss;
		fc.fbase = 0;
		fc.fscope = 0;

		/* If the base of the search is missing, signal a refresh */
		rc = syncprov_findbase( op, &fc );
		if ( rc != LDAP_SUCCESS ) {
			SlapReply rs = {REP_RESULT};
			send_ldap_error( ss->s_op, &rs, LDAP_SYNC_REFRESH_REQUIRED,
				"search base has changed" );
			sprev->s_next = snext;
			syncprov_drop_psearch( ss, 1 );
			ss = sprev;
			continue;
		}


		/* If we're sending results now, look for this op in old matches */
		if ( !saveit ) {
			syncmatches *old;

			/* Did we modify the search base? */
			if ( dn_match( &op->o_req_ndn, &ss->s_base )) {
				ldap_pvt_thread_mutex_lock( &ss->s_mutex );
				ss->s_flags |= PS_WROTE_BASE;
				ldap_pvt_thread_mutex_unlock( &ss->s_mutex );
			}

			for ( sm=opc->smatches, old=(syncmatches *)&opc->smatches; sm;
				old=sm, sm=sm->sm_next ) {
				if ( sm->sm_op == ss ) {
					found = 1;
					old->sm_next = sm->sm_next;
					op->o_tmpfree( sm, op->o_tmpmemctx );
					break;
				}
			}
		}

		/* check if current o_req_dn is in scope and matches filter */
		if ( fc.fscope && test_filter( op, e, ss->s_op->ors_filter ) ==
			LDAP_COMPARE_TRUE ) {
			if ( saveit ) {
				sm = op->o_tmpalloc( sizeof(syncmatches), op->o_tmpmemctx );
				sm->sm_next = opc->smatches;
				sm->sm_op = ss;
				ldap_pvt_thread_mutex_lock( &ss->s_mutex );
				++ss->s_inuse;
				ldap_pvt_thread_mutex_unlock( &ss->s_mutex );
				opc->smatches = sm;
			} else {
				/* if found send UPDATE else send ADD */
				syncprov_qresp( opc, ss,
					found ? LDAP_SYNC_MODIFY : LDAP_SYNC_ADD );
			}
		} else if ( !saveit && found ) {
			/* send DELETE */
			syncprov_qresp( opc, ss, LDAP_SYNC_DELETE );
		}
	}
	ldap_pvt_thread_mutex_unlock( &si->si_ops_mutex );

	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;
	}
	if ( freefdn ) {
		op->o_tmpfree( fc.fdn->bv_val, op->o_tmpmemctx );
	}
	op->o_bd = b0;
}

static int
syncprov_op_cleanup( Operation *op, SlapReply *rs )
{
	slap_callback *cb = op->o_callback;
	opcookie *opc = cb->sc_private;
	slap_overinst *on = opc->son;
	syncprov_info_t		*si = on->on_bi.bi_private;
	syncmatches *sm, *snext;
	modtarget *mt, mtdummy;

	for (sm = opc->smatches; sm; sm=snext) {
		snext = sm->sm_next;
		syncprov_free_syncop( sm->sm_op );
		op->o_tmpfree( sm, op->o_tmpmemctx );
	}

	/* Remove op from lock table */
	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 ) {
		modinst *mi = mt->mt_mods;

		/* If there are more, promote the next one */
		ldap_pvt_thread_mutex_lock( &mt->mt_mutex );
		if ( mi->mi_next ) {
			mt->mt_mods = mi->mi_next;
			mt->mt_op = mt->mt_mods->mi_op;
			ldap_pvt_thread_mutex_unlock( &mt->mt_mutex );
		} else {
			avl_delete( &si->si_mods, mt, sp_avl_cmp );
			ldap_pvt_thread_mutex_unlock( &mt->mt_mutex );
			ldap_pvt_thread_mutex_destroy( &mt->mt_mutex );
			ch_free( mt );
		}
	}
	ldap_pvt_thread_mutex_unlock( &si->si_mods_mutex );
	if ( !BER_BVISNULL( &opc->suuid ))
		op->o_tmpfree( opc->suuid.bv_val, op->o_tmpmemctx );
	if ( !BER_BVISNULL( &opc->sndn ))
		op->o_tmpfree( opc->sndn.bv_val, op->o_tmpmemctx );
	if ( !BER_BVISNULL( &opc->sdn ))
		op->o_tmpfree( opc->sdn.bv_val, op->o_tmpmemctx );
	op->o_callback = cb->sc_next;
	op->o_tmpfree(cb, op->o_tmpmemctx);

	return 0;
}

static void
syncprov_checkpoint( Operation *op, SlapReply *rs, slap_overinst *on )
{
	syncprov_info_t		*si = on->on_bi.bi_private;
	Modifications mod;
	Operation opm;
	SlapReply rsm = { 0 };
	struct berval bv[2];
	slap_callback cb = {0};

	/* 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;
	mod.sml_flags = 0;
	mod.sml_next = NULL;

	cb.sc_response = slap_null_cb;
	opm = *op;
	opm.o_tag = LDAP_REQ_MODIFY;
	opm.o_callback = &cb;
	opm.orm_modlist = &mod;
	opm.o_req_dn = op->o_bd->be_suffix[0];
	opm.o_req_ndn = op->o_bd->be_nsuffix[0];
	opm.o_bd->bd_info = on->on_info->oi_orig;
	opm.o_managedsait = SLAP_CONTROL_NONCRITICAL;
	SLAP_DBFLAGS( opm.o_bd ) |= SLAP_DBFLAG_NOLASTMOD;
	opm.o_bd->be_modify( &opm, &rsm );
	SLAP_DBFLAGS( opm.o_bd ) ^= SLAP_DBFLAG_NOLASTMOD;
	if ( mod.sml_next != NULL ) {
		slap_mods_free( mod.sml_next, 1 );
	}
}

static void
syncprov_add_slog( Operation *op )
{
	opcookie *opc = op->o_callback->sc_private;
	slap_overinst *on = opc->son;
	syncprov_info_t		*si = on->on_bi.bi_private;
	sessionlog *sl;
	slog_entry *se;

	sl = si->si_logs;
	{
		/* Allocate a record. UUIDs are not NUL-terminated. */
		se = ch_malloc( sizeof( slog_entry ) + opc->suuid.bv_len + 
			op->o_csn.bv_len + 1 );
		se->se_next = NULL;
		se->se_tag = op->o_tag;

		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, 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_tail->se_next = se;
		} else {
			sl->sl_head = se;
		}
		sl->sl_tail = se;
		sl->sl_num++;
		while ( sl->sl_num > sl->sl_size ) {
			se = sl->sl_head;
			sl->sl_head = se->se_next;
			strcpy( sl->sl_mincsn.bv_val, se->se_csn.bv_val );
			sl->sl_mincsn.bv_len = se->se_csn.bv_len;
			ch_free( se );
			sl->sl_num--;
		}
		ldap_pvt_thread_mutex_unlock( &sl->sl_mutex );
	}
}

/* Just set a flag if we found the matching entry */
static int
playlog_cb( Operation *op, SlapReply *rs )
{
	if ( rs->sr_type == REP_SEARCH ) {
		op->o_callback->sc_private = (void *)1;
	}
	return rs->sr_err;
}

/* enter with sl->sl_mutex locked, release before returning */
static void
syncprov_playlog( Operation *op, SlapReply *rs, sessionlog *sl,
	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 );
		return;
	}

	num = sl->sl_num;
	i = 0;
	nmods = 0;

	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.
	 */
	Debug( LDAP_DEBUG_SYNC, "srs csn %s\n", srs->sr_state.ctxcsn.bv_val, 0, 0 );
	for ( se=sl->sl_head; se; se=se->se_next ) {
		Debug( LDAP_DEBUG_SYNC, "log csn %s\n", se->se_csn.bv_val, 0, 0 );
		ndel = ber_bvcmp( &se->se_csn, &srs->sr_state.ctxcsn );
		if ( ndel <= 0 ) {
			Debug( LDAP_DEBUG_SYNC, "cmp %d, too old\n", ndel, 0, 0 );
			continue;
		}
		ndel = ber_bvcmp( &se->se_csn, ctxcsn );
		if ( ndel > 0 ) {
			Debug( LDAP_DEBUG_SYNC, "cmp %d, too new\n", ndel, 0, 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;
		}
		uuids[j].bv_val = uuids[0].bv_val + (j * UUID_LEN);
		AC_MEMCPY(uuids[j].bv_val, se->se_uuid.bv_val, UUID_LEN);
		uuids[j].bv_len = UUID_LEN;
	}
	ldap_pvt_thread_mutex_unlock( &sl->sl_mutex );

	ndel = i;

	/* Zero out unused slots */
	for ( i=ndel; i < num - nmods; i++ )
		uuids[i].bv_len = 0;

	/* Mods must be validated to see if they belong in this delete set.
	 */

	mmods = nmods;
	/* Strip any duplicates */
	for ( i=0; i<nmods; i++ ) {
		for ( j=0; j<ndel; j++ ) {
			if ( bvmatch( &uuids[j], &uuids[num - 1 - i] )) {
				uuids[num - 1 - i].bv_len = 0;
				mmods --;
				break;
			}
		}
		if ( uuids[num - 1 - i].bv_len == 0 ) continue;
		for ( j=0; j<i; j++ ) {
			if ( bvmatch( &uuids[num - 1 - j], &uuids[num - 1 - i] )) {
				uuids[num - 1 - i].bv_len = 0;
				mmods --;
				break;
			}
		}
	}

	if ( mmods ) {
		Operation fop;
		SlapReply frs = { REP_RESULT };
		int rc;
		Filter mf, af;
#ifdef LDAP_COMP_MATCH
		AttributeAssertion eq = { NULL, BER_BVNULL, NULL };
#else
		AttributeAssertion eq;
#endif
		slap_callback cb = {0};

		fop = *op;

		fop.o_sync_mode = 0;
		fop.o_callback = &cb;
		fop.ors_limit = NULL;
		fop.ors_tlimit = SLAP_NO_LIMIT;
		fop.ors_attrs = slap_anlist_all_attributes;