syncprov.c

ldap服务器源码

syncprov_op_search( Operation *op, SlapReply *rs )
{
	slap_overinst		*on = (slap_overinst *)op->o_bd->bd_info;
	syncprov_info_t		*si = (syncprov_info_t *)on->on_bi.bi_private;
	slap_callback	*cb;
	int gotstate = 0, nochange = 0, do_present;
	syncops *sop = NULL;
	searchstate *ss;
	sync_control *srs;
	struct berval ctxcsn;
	char csnbuf[LDAP_LUTIL_CSNSTR_BUFSIZE];

	if ( !(op->o_sync_mode & SLAP_SYNC_REFRESH) ) return SLAP_CB_CONTINUE;

	if ( op->ors_deref & LDAP_DEREF_SEARCHING ) {
		send_ldap_error( op, rs, LDAP_PROTOCOL_ERROR, "illegal value for derefAliases" );
		return rs->sr_err;
	}

	do_present = si->si_nopres ? 0 : 1;

	srs = op->o_controls[slap_cids.sc_LDAPsync];
	op->o_managedsait = SLAP_CONTROL_NONCRITICAL;

	/* If this is a persistent search, set it up right away */
	if ( op->o_sync_mode & SLAP_SYNC_PERSIST ) {
		syncops so = {0};
		fbase_cookie fc;
		opcookie opc;
		slap_callback sc;

		fc.fss = &so;
		fc.fbase = 0;
		so.s_eid = NOID;
		so.s_op = op;
		so.s_flags = PS_IS_REFRESHING | PS_FIND_BASE;
		/* 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 );
			return rs->sr_err;
		}
		sop = ch_malloc( sizeof( syncops ));
		*sop = so;
		ldap_pvt_thread_mutex_init( &sop->s_mutex );
		sop->s_rid = srs->sr_state.rid;
		sop->s_inuse = 1;

		ldap_pvt_thread_mutex_lock( &si->si_ops_mutex );
		sop->s_next = si->si_ops;
		si->si_ops = sop;
		ldap_pvt_thread_mutex_unlock( &si->si_ops_mutex );
	}

	/* snapshot the ctxcsn */
	ldap_pvt_thread_rdwr_rlock( &si->si_csn_rwlock );
	strcpy( csnbuf, si->si_ctxcsnbuf );
	ctxcsn.bv_len = si->si_ctxcsn.bv_len;
	ldap_pvt_thread_rdwr_runlock( &si->si_csn_rwlock );
	ctxcsn.bv_val = csnbuf;
	
	/* If we have a cookie, handle the PRESENT lookups */
	if ( !BER_BVISNULL( &srs->sr_state.ctxcsn )) {
		sessionlog *sl;

		/* The cookie was validated when it was parsed, just use it */

		/* If just Refreshing and nothing has changed, shortcut it */
		if ( bvmatch( &srs->sr_state.ctxcsn, &ctxcsn )) {
			nochange = 1;
			if ( !(op->o_sync_mode & SLAP_SYNC_PERSIST) ) {
				LDAPControl	*ctrls[2];

				ctrls[0] = NULL;
				ctrls[1] = NULL;
				syncprov_done_ctrl( op, rs, ctrls, 0, 0,
					NULL, LDAP_SYNC_REFRESH_DELETES );
				rs->sr_ctrls = ctrls;
				rs->sr_err = LDAP_SUCCESS;
				send_ldap_result( op, rs );
				rs->sr_ctrls = NULL;
				return rs->sr_err;
			}
			goto shortcut;
		}
		/* Do we have a sessionlog for this search? */
		sl=si->si_logs;
		if ( sl ) {
			ldap_pvt_thread_mutex_lock( &sl->sl_mutex );
			/* 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, &ctxcsn );
			} else {
				ldap_pvt_thread_mutex_unlock( &sl->sl_mutex );
			}
		}
		/* Is the CSN still present in the database? */
		if ( syncprov_findcsn( op, FIND_CSN ) != LDAP_SUCCESS ) {
			/* No, so a reload is required */
			/* the 2.2 consumer doesn't send this hint */
			if ( si->si_usehint && srs->sr_rhint == 0 ) {
				send_ldap_error( op, rs, LDAP_SYNC_REFRESH_REQUIRED, "sync cookie is stale" );
				return rs->sr_err;
			}
		} else {
			gotstate = 1;
			/* If changed and doing Present lookup, send Present UUIDs */
			if ( do_present && syncprov_findcsn( op, FIND_PRESENT ) !=
				LDAP_SUCCESS ) {
				send_ldap_result( op, rs );
				return rs->sr_err;
			}
		}
	}

shortcut:
	/* Append CSN range to search filter, save original filter
	 * for persistent search evaluation
	 */
	if ( sop ) {
		sop->s_filterstr= op->ors_filterstr;
	}

	/* If something changed, find the changes */
	if ( gotstate && !nochange ) {
		Filter *fand, *fava;

		fand = op->o_tmpalloc( sizeof(Filter), op->o_tmpmemctx );
		fand->f_choice = LDAP_FILTER_AND;
		fand->f_next = NULL;
		fava = op->o_tmpalloc( sizeof(Filter), op->o_tmpmemctx );
		fand->f_and = fava;
		fava->f_choice = LDAP_FILTER_GE;
		fava->f_ava = op->o_tmpalloc( sizeof(AttributeAssertion), op->o_tmpmemctx );
		fava->f_ava->aa_desc = slap_schema.si_ad_entryCSN;
#ifdef LDAP_COMP_MATCH
		fava->f_ava->aa_cf = NULL;
#endif
		ber_dupbv_x( &fava->f_ava->aa_value, &srs->sr_state.ctxcsn, 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 */
	cb = op->o_tmpcalloc(1, sizeof(slap_callback)+sizeof(searchstate), op->o_tmpmemctx);
	ss = (searchstate *)(cb+1);
	ss->ss_on = on;
	ss->ss_so = sop;
	ss->ss_present = do_present;
	ss->ss_ctxcsn.bv_len = ctxcsn.bv_len;
	ss->ss_ctxcsn.bv_val = ss->ss_csnbuf;
	strcpy( ss->ss_ctxcsn.bv_val, ctxcsn.bv_val );
	cb->sc_response = syncprov_search_response;
	cb->sc_cleanup = syncprov_search_cleanup;
	cb->sc_private = ss;
	cb->sc_next = op->o_callback;
	op->o_callback = cb;

#if 0	/* I don't think we need to shortcircuit back-bdb any more */
	op->o_sync_mode &= SLAP_CONTROL_MASK;
#endif

	/* If this is a persistent search and no changes were reported during
	 * the refresh phase, just invoke the response callback to transition
	 * us into persist phase
	 */
	if ( nochange ) {
		rs->sr_err = LDAP_SUCCESS;
		rs->sr_nentries = 0;
		send_ldap_result( op, rs );
		return rs->sr_err;
	}
	return SLAP_CB_CONTINUE;
}

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

	if ( rs->sr_entry &&
		dn_match( &rs->sr_entry->e_nname, op->o_bd->be_nsuffix )) {

		if ( SLAP_OPATTRS( rs->sr_attr_flags ) ||
			ad_inlist( slap_schema.si_ad_contextCSN, rs->sr_attrs )) {
			Attribute *a, **ap = NULL;

			for ( a=rs->sr_entry->e_attrs; a; a=a->a_next ) {
				if ( a->a_desc == slap_schema.si_ad_contextCSN )
					break;
			}

			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->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;
			}

			ldap_pvt_thread_rdwr_rlock( &si->si_csn_rwlock );
			if ( !ap ) {
				strcpy( a->a_vals[0].bv_val, si->si_ctxcsnbuf );
			} else {
				ber_dupbv( &a->a_vals[0], &si->si_ctxcsn );
			}
			ldap_pvt_thread_rdwr_runlock( &si->si_csn_rwlock );
		}
	}
	return SLAP_CB_CONTINUE;
}

enum {
	SP_CHKPT = 1,
	SP_SESSL,
	SP_NOPRES,
	SP_USEHINT
};

static ConfigDriver sp_cf_gen;

static ConfigTable spcfg[] = {
	{ "syncprov-checkpoint", "ops> <minutes", 3, 3, 0, ARG_MAGIC|SP_CHKPT,
		sp_cf_gen, "( OLcfgOvAt:1.1 NAME 'olcSpCheckpoint' "
			"DESC 'ContextCSN checkpoint interval in ops and minutes' "
			"SYNTAX OMsDirectoryString SINGLE-VALUE )", NULL, NULL },
	{ "syncprov-sessionlog", "ops", 2, 2, 0, ARG_INT|ARG_MAGIC|SP_SESSL,
		sp_cf_gen, "( OLcfgOvAt:1.2 NAME 'olcSpSessionlog' "
			"DESC 'Session log size in ops' "
			"SYNTAX OMsInteger SINGLE-VALUE )", NULL, NULL },
	{ "syncprov-nopresent", NULL, 2, 2, 0, ARG_ON_OFF|ARG_MAGIC|SP_NOPRES,
		sp_cf_gen, "( OLcfgOvAt:1.3 NAME 'olcSpNoPresent' "
			"DESC 'Omit Present phase processing' "
			"SYNTAX OMsBoolean SINGLE-VALUE )", NULL, NULL },
	{ "syncprov-reloadhint", NULL, 2, 2, 0, ARG_ON_OFF|ARG_MAGIC|SP_USEHINT,
		sp_cf_gen, "( OLcfgOvAt:1.4 NAME 'olcSpReloadHint' "
			"DESC 'Observe Reload Hint in Request control' "
			"SYNTAX OMsBoolean SINGLE-VALUE )", NULL, NULL },
	{ NULL, NULL, 0, 0, 0, ARG_IGNORED }
};

static ConfigOCs spocs[] = {
	{ "( OLcfgOvOc:1.1 "
		"NAME 'olcSyncProvConfig' "
		"DESC 'SyncRepl Provider configuration' "
		"SUP olcOverlayConfig "
		"MAY ( olcSpCheckpoint $ olcSpSessionlog $ olcSpNoPresent ) )",
			Cft_Overlay, spcfg },
	{ NULL, 0, NULL }
};

static int
sp_cf_gen(ConfigArgs *c)
{
	slap_overinst		*on = (slap_overinst *)c->bi;
	syncprov_info_t		*si = (syncprov_info_t *)on->on_bi.bi_private;
	int rc = 0;

	if ( c->op == SLAP_CONFIG_EMIT ) {
		switch ( c->type ) {
		case SP_CHKPT:
			if ( si->si_chkops || si->si_chktime ) {
				struct berval bv;
				bv.bv_len = sprintf( c->msg, "%d %d",
					si->si_chkops, si->si_chktime );
				bv.bv_val = c->msg;
				value_add_one( &c->rvalue_vals, &bv );
			} else {
				rc = 1;
			}
			break;
		case SP_SESSL:
			if ( si->si_logs ) {
				c->value_int = si->si_logs->sl_size;
			} else {
				rc = 1;
			}
			break;
		case SP_NOPRES:
			if ( si->si_nopres ) {
				c->value_int = 1;
			} else {
				rc = 1;
			}
			break;
		case SP_USEHINT:
			if ( si->si_usehint ) {
				c->value_int = 1;
			} else {
				rc = 1;
			}
			break;
		}
		return rc;
	} else if ( c->op == LDAP_MOD_DELETE ) {
		switch ( c->type ) {
		case SP_CHKPT:
			si->si_chkops = 0;
			si->si_chktime = 0;
			break;
		case SP_SESSL:
			if ( si->si_logs )
				si->si_logs->sl_size = 0;
			else
				rc = LDAP_NO_SUCH_ATTRIBUTE;
			break;
		case SP_NOPRES:
			if ( si->si_nopres )
				si->si_nopres = 0;
			else
				rc = LDAP_NO_SUCH_ATTRIBUTE;
			break;
		case SP_USEHINT:
			if ( si->si_usehint )
				si->si_usehint = 0;
			else
				rc = LDAP_NO_SUCH_ATTRIBUTE;
			break;
		}
		return rc;
	}
	switch ( c->type ) {
	case SP_CHKPT:
		if ( lutil_atoi( &si->si_chkops, c->argv[1] ) != 0 ) {
			snprintf( c->msg, sizeof( c->msg ), "%s unable to parse checkpoint ops # \"%s\"",
				c->argv[0], c->argv[1] );
			Debug( LDAP_DEBUG_CONFIG|LDAP_DEBUG_NONE,
				"%s: %s\n", c->log, c->msg, 0 );
			return ARG_BAD_CONF;
		}
		if ( si->si_chkops <= 0 ) {
			snprintf( c->msg, sizeof( c->msg ), "%s invalid checkpoint ops # \"%d\"",
				c->argv[0], si->si_chkops );
			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 ) {
			snprintf( c->msg, sizeof( c->msg ), "%s unable to parse checkpoint time \"%s\"",
				c->argv[0], c->argv[1] );
			Debug( LDAP_DEBUG_CONFIG|LDAP_DEBUG_NONE,
				"%s: %s\n", c->log, c->msg, 0 );
			return ARG_BAD_CONF;
		}
		if ( si->si_chktime <= 0 ) {
			snprintf( c->msg, sizeof( c->msg ), "%s invalid checkpoint time \"%d\"",
				c->argv[0], si->si_chkops );
			Debug( LDAP_DEBUG_CONFIG|LDAP_DEBUG_NONE,
				"%s: %s\n", c->log, c->msg, 0 );
			return ARG_BAD_CONF;
		}
		si->si_chktime *= 60;
		break;
	case SP_SESSL: {
		sessionlog *sl;
		int size = c->value_int;

		if ( size < 0 ) {
			snprintf( c->msg, sizeof( c->msg ), "%s size %d is negative",
				c->argv[0], size );
			Debug( LDAP_DEBUG_CONFIG|LDAP_DEBUG_NONE,
				"%s: %s\n", c->log, c->msg, 0 );
			return ARG_BAD_CONF;
		}
		sl = si->si_logs;
		if ( !sl ) {
			sl = ch_malloc( sizeof( sessionlog ) + LDAP_LUTIL_CSNSTR_BUFSIZE );
			sl->sl_mincsn.bv_val = (char *)(sl+1);
			sl->sl_mincsn.bv_len = 0;
			sl->sl_num = 0;
			sl->sl_head = sl->sl_tail = NULL;
			ldap_pvt_thread_mutex_init( &sl->sl_mutex );
			si->si_logs = sl;
		}
		sl->sl_size = size;
		}
		break;
	case SP_NOPRES:
		si->si_nopres = c->value_int;
		break;
	case SP_USEHINT:
		si->si_usehint = c->value_int;
		break;
	}
	return rc;
}

/* ITS#3456 we cannot run this search on the main thread, must use a
 * child thread in order to insure we have a big enough stack.
 */
static void *
syncprov_db_otask(
	void *ptr
)
{
	syncprov_findcsn( ptr, FIND_MAXCSN );
	return NULL;
}

/* Read any existing contextCSN from the underlying db.
 * Then search for any entries newer than that. If no value exists,
 * just generate it. Cache whatever result.
 */
static int
syncprov_db_open(
    BackendDB *be
)
{
	slap_overinst   *on = (slap_overinst *) be->bd_info;
	syncprov_info_t *si = (syncprov_info_t *)on->on_bi.bi_private;

	Connection conn = { 0 };
	OperationBuffer opbuf = { 0 };
	char ctxcsnbuf[LDAP_LUTIL_CSNSTR_BUFSIZE];
	Operation *op = (Operation *) &opbuf;
	Entry *e;
	Attribute *a;
	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;
	}

	rc = overlay_register_control( be, LDAP_CONTROL_SYNC );
	if ( rc ) {
		return rc;
	}

	thrctx = ldap_pvt_thread_pool_context();
	connection_fake_init( &conn, op, thrctx );
	op->o_bd = be;
	op->o_dn = be->be_rootdn;
	op->o_ndn = be->be_rootndn;

	ctxcsnbuf[0] = '\0';

	op->o_bd->bd_info = on->on_info->oi_orig;
	rc = be_entry_get_rw( op, be->be_nsuffix, NULL,
		slap_schema.si_ad_contextCSN, 0, &e );

	if ( e ) {
		ldap_pvt_thread_t tid;

		a = attr_find( e->e_attrs, slap_schema.si_ad_contextCSN );
		if ( a ) {
			si->si_ctxcsn.bv_len = a->a_nvals[0].bv_len;
			if ( si->si_ctxcsn.bv_len >= sizeof(si->si_ctxcsnbuf ))
				si->si_ctxcsn.bv_len = sizeof(si->si_ctxcsnbuf)-1;
			strncpy( si->si_ctxcsnbuf, a->a_nvals[0].bv_val,
				si->si_ctxcsn.bv_len );
			si->si_ctxcsnbuf[si->si_ctxcsn.bv_len] = '\0';
			strcpy( ctxcsnbuf, si->si_