pcache.c

ldap服务器源码

	} else if ( rs->sr_type == REP_RESULT ) {
		QueryTemplate* templ = (qm->templates)+si->template_id;
		if (( si->count && cache_entries( op, rs, &uuid ) == 0 ) ||
			( templ->negttl && !si->count && !si->over &&
				rs->sr_err == LDAP_SUCCESS )) {
			qm->addfunc(qm, &si->query, si->template_id,
				si->count ? &uuid : NULL);

			ldap_pvt_thread_mutex_lock(&cm->cache_mutex);
			cm->num_cached_queries++;
			Debug( pcache_debug, "STORED QUERIES = %lu\n",
					cm->num_cached_queries, 0, 0 );
			ldap_pvt_thread_mutex_unlock(&cm->cache_mutex);

			/* If the consistency checker suspended itself,
			 * wake it back up
			 */
			if ( cm->cc_paused ) {
				ldap_pvt_thread_mutex_lock( &slapd_rq.rq_mutex );
				if ( cm->cc_paused ) {
					cm->cc_paused = 0;
					ldap_pvt_runqueue_resched( &slapd_rq, cm->cc_arg, 0 );
				}
				ldap_pvt_thread_mutex_unlock( &slapd_rq.rq_mutex );
			}
		} else {
			free( si->query.attrs );
			filter_free( si->query.filter );
		}

		/* free self */
		op->o_callback->sc_cleanup = slap_freeself_cb;
	}
	return SLAP_CB_CONTINUE;
}

static void
add_filter_attrs(
	Operation *op,
	AttributeName** new_attrs,
	struct attr_set *attrs,
	AttributeName* filter_attrs,
	int fattr_cnt,
	int fattr_got_oc)
{
	int alluser = 0;
	int allop = 0;
	int i, j;
	int count;
	int addoc = 0;

	/* duplicate attrs */
	count = attrs->count + fattr_cnt;
	if ( !fattr_got_oc && !(attrs->flags & PC_GOT_OC)) {
		addoc = 1;
		count++;
	}

	*new_attrs = (AttributeName*)ch_malloc((count+1)*
		sizeof(AttributeName));
	for (i=0; i<attrs->count; i++) {
		(*new_attrs)[i].an_name = attrs->attrs[i].an_name;
		(*new_attrs)[i].an_desc = attrs->attrs[i].an_desc;
	}
	BER_BVZERO( &(*new_attrs)[i].an_name );
	alluser = an_find(*new_attrs, &AllUser);
	allop = an_find(*new_attrs, &AllOper);

	j = i;
	for ( i=0; i<fattr_cnt; i++ ) {
		if ( an_find(*new_attrs, &filter_attrs[i].an_name ))
			continue;
		if ( is_at_operational(filter_attrs[i].an_desc->ad_type) ) {
			if (allop)
				continue;
		} else if (alluser)
			continue;
		(*new_attrs)[j].an_name = filter_attrs[i].an_name;
		(*new_attrs)[j].an_desc = filter_attrs[i].an_desc;
		(*new_attrs)[j].an_oc = NULL;
		(*new_attrs)[j].an_oc_exclude = 0;
		j++;
	}
	if ( addoc ) {
		(*new_attrs)[j].an_name = slap_schema.si_ad_objectClass->ad_cname;
		(*new_attrs)[j].an_desc = slap_schema.si_ad_objectClass;
		(*new_attrs)[j].an_oc = NULL;
		(*new_attrs)[j].an_oc_exclude = 0;
		j++;
	}
	BER_BVZERO( &(*new_attrs)[j].an_name );
}

/* NOTE: this is a quick workaround to let pcache minimally interact
 * with pagedResults.  A more articulated solutions would be to
 * perform the remote query without control and cache all results,
 * performing the pagedResults search only within the client
 * and the proxy.  This requires pcache to understand pagedResults. */
static int
pcache_chk_controls(
	Operation	*op,
	SlapReply	*rs )
{
	const char	*non = "";
	const char	*stripped = "";

	switch( op->o_pagedresults ) {
	case SLAP_CONTROL_NONCRITICAL:
		non = "non-";
		stripped = "; stripped";
		/* fallthru */

	case SLAP_CONTROL_CRITICAL:
		Debug( pcache_debug, "%s: "
			"%scritical pagedResults control "
			"disabled with proxy cache%s.\n",
			op->o_log_prefix, non, stripped );
		
		slap_remove_control( op, rs, slap_cids.sc_pagedResults, NULL );
		break;

	default:
		rs->sr_err = SLAP_CB_CONTINUE;
		break;
	}

	return rs->sr_err;
}

static int
pcache_op_search(
	Operation	*op,
	SlapReply	*rs )
{
	slap_overinst *on = (slap_overinst *)op->o_bd->bd_info;
	cache_manager *cm = on->on_bi.bi_private;
	query_manager*		qm = cm->qm;

	int count;

	int i = -1;

	AttributeName	*filter_attrs = NULL;

	Query		query;

	int 		attr_set = -1;
	int 		template_id = -1;
	CachedQuery 	*answerable = NULL;
	int 		cacheable = 0;
	int		fattr_cnt=0;
	int		fattr_got_oc = 0;
	int		oc_attr_absent = 1;

	struct berval tempstr;

	tempstr.bv_val = op->o_tmpalloc( op->ors_filterstr.bv_len+1, op->o_tmpmemctx );
	tempstr.bv_len = 0;
	if ( filter2template( op, op->ors_filter, &tempstr, &filter_attrs,
		&fattr_cnt, &fattr_got_oc )) {
		op->o_tmpfree( tempstr.bv_val, op->o_tmpmemctx );
		return SLAP_CB_CONTINUE;
	}

	Debug( pcache_debug, "query template of incoming query = %s\n",
					tempstr.bv_val, 0, 0 );

	/* FIXME: cannot cache/answer requests with pagedResults control */
	

	/* find attr set */
	attr_set = get_attr_set(op->ors_attrs, qm, cm->numattrsets);

	query.filter = op->ors_filter;
	query.attrs = op->ors_attrs;
	query.save_attrs = NULL;
	query.base = op->o_req_ndn;
	query.scope = op->ors_scope;

	/* check for query containment */
	if (attr_set > -1) {
		QueryTemplate *qt = qm->templates;
		for (i=0; i<cm->numtemplates; i++, qt++) {
			/* find if template i can potentially answer tempstr */
			if ( qt->attr_set_index != attr_set ||
				qt->querystr.bv_len != tempstr.bv_len ||
				strcasecmp( qt->querystr.bv_val, tempstr.bv_val ))
				continue;
			cacheable = 1;
			template_id = i;
			Debug( LDAP_DEBUG_NONE, "Entering QC, querystr = %s\n",
			 		op->ors_filterstr.bv_val, 0, 0 );
			answerable = (*(qm->qcfunc))(op, qm, &query, i);

			if (answerable)
				break;
		}
	}
	op->o_tmpfree( tempstr.bv_val, op->o_tmpmemctx );

	query.save_attrs = op->ors_attrs;
	query.attrs = NULL;

	if (answerable) {
		/* Need to clear the callbacks of the original operation,
		 * in case there are other overlays */
		BackendDB	*save_bd = op->o_bd;
		slap_callback	*save_cb = op->o_callback;

		Debug( pcache_debug, "QUERY ANSWERABLE\n", 0, 0, 0 );
		op->o_tmpfree( filter_attrs, op->o_tmpmemctx );
		ldap_pvt_thread_rdwr_runlock(&qm->templates[i].t_rwlock);
		if ( BER_BVISNULL( &answerable->q_uuid )) {
			/* No entries cached, just an empty result set */
			i = rs->sr_err = 0;
			send_ldap_result( op, rs );
		} else {
			op->o_bd = &cm->db;
			op->o_callback = NULL;
			i = cm->db.bd_info->bi_op_search( op, rs );
		}
		op->o_bd = save_bd;
		op->o_callback = save_cb;
		return i;
	}

	Debug( pcache_debug, "QUERY NOT ANSWERABLE\n", 0, 0, 0 );

	ldap_pvt_thread_mutex_lock(&cm->cache_mutex);
	if (cm->num_cached_queries >= cm->max_queries) {
		cacheable = 0;
	}
	ldap_pvt_thread_mutex_unlock(&cm->cache_mutex);

	if (op->ors_attrsonly)
		cacheable = 0;

	if (cacheable) {
		slap_callback		*cb;
		struct search_info	*si;

		Debug( pcache_debug, "QUERY CACHEABLE\n", 0, 0, 0 );
		query.filter = filter_dup(op->ors_filter, NULL);
		add_filter_attrs(op, &query.attrs, &qm->attr_sets[attr_set],
			filter_attrs, fattr_cnt, fattr_got_oc);

		op->ors_attrs = query.attrs;

		cb = op->o_tmpalloc( sizeof(*cb) + sizeof(*si), op->o_tmpmemctx);
		cb->sc_response = pcache_response;
		cb->sc_cleanup = NULL;
		cb->sc_private = (cb+1);
		si = cb->sc_private;
		si->on = on;
		si->query = query;
		si->template_id = template_id;
		si->max = cm->num_entries_limit ;
		si->over = 0;
		si->count = 0;
		si->head = NULL;
		si->tail = NULL;

		if ( cm->response_cb == PCACHE_RESPONSE_CB_HEAD ) {
			cb->sc_next = op->o_callback;
			op->o_callback = cb;

		} else {
			slap_callback		**pcb;

			/* need to move the callback at the end, in case other
			 * overlays are present, so that the final entry is
			 * actually cached */
			cb->sc_next = NULL;
			for ( pcb = &op->o_callback; *pcb; pcb = &(*pcb)->sc_next );
			*pcb = cb;
		}

	} else {
		Debug( pcache_debug, "QUERY NOT CACHEABLE\n",
					0, 0, 0);
	}

	op->o_tmpfree( filter_attrs, op->o_tmpmemctx );

	return SLAP_CB_CONTINUE;
}

static int
get_attr_set(
	AttributeName* attrs,
	query_manager* qm,
	int num )
{
	int i;
	int count = 0;

	if ( attrs ) {
		for ( ; attrs[count].an_name.bv_val; count++ );
	}

	for (i=0; i<num; i++) {
		AttributeName *a2;
		int found = 1;

		if ( count > qm->attr_sets[i].count )
			continue;
		if ( !count ) {
			if ( !qm->attr_sets[i].count )
				break;
			continue;
		}
		for ( a2 = attrs; a2->an_name.bv_val; a2++ ) {
			if ( !an_find( qm->attr_sets[i].attrs, &a2->an_name )) {
				found = 0;
				break;
			}
		}
		if ( found )
			break;
	}
	if ( i == num )
		i = -1;
	return i;
}

static void*
consistency_check(
	void *ctx,
	void *arg )
{
	struct re_s *rtask = arg;
	slap_overinst *on = rtask->arg;
	cache_manager *cm = on->on_bi.bi_private;
	query_manager *qm = cm->qm;
	Connection conn = {0};
	OperationBuffer opbuf;
	Operation *op;

	SlapReply rs = {REP_RESULT};
	CachedQuery* query, *query_prev;
	int i, return_val, pause = 1;
	QueryTemplate* templ;

	op = (Operation *) &opbuf;
	connection_fake_init( &conn, op, ctx );

	op->o_bd = &cm->db;
	op->o_dn = cm->db.be_rootdn;
	op->o_ndn = cm->db.be_rootndn;

      	cm->cc_arg = arg;

	for (i=0; qm->templates[i].querystr.bv_val; i++) {
		templ = qm->templates + i;
		query = templ->query_last;
		if ( query ) pause = 0;
		op->o_time = slap_get_time();
		while (query && (query->expiry_time < op->o_time)) {
			Debug( pcache_debug, "Lock CR index = %d\n",
					i, 0, 0 );
			ldap_pvt_thread_rdwr_wlock(&templ->t_rwlock);
			remove_from_template(query, templ);
			Debug( pcache_debug, "TEMPLATE %d QUERIES-- %d\n",
					i, templ->no_of_queries, 0 );
			Debug( pcache_debug, "Unlock CR index = %d\n",
					i, 0, 0 );
			ldap_pvt_thread_rdwr_wunlock(&templ->t_rwlock);
			ldap_pvt_thread_mutex_lock(&qm->lru_mutex);
			remove_query(qm, query);
			ldap_pvt_thread_mutex_unlock(&qm->lru_mutex);
			if ( BER_BVISNULL( &query->q_uuid ))
				return_val = 0;
			else
				return_val = remove_query_data(op, &rs, &query->q_uuid);
			Debug( pcache_debug, "STALE QUERY REMOVED, SIZE=%d\n",
						return_val, 0, 0 );
			ldap_pvt_thread_mutex_lock(&cm->cache_mutex);
			cm->cur_entries -= return_val;
			cm->num_cached_queries--;
			Debug( pcache_debug, "STORED QUERIES = %lu\n",
					cm->num_cached_queries, 0, 0 );
			ldap_pvt_thread_mutex_unlock(&cm->cache_mutex);
			Debug( pcache_debug,
				"STALE QUERY REMOVED, CACHE ="
				"%d entries\n",
				cm->cur_entries, 0, 0 );
			query_prev = query;
			query = query->prev;
			free_query(query_prev);
		}
	}
	ldap_pvt_thread_mutex_lock( &slapd_rq.rq_mutex );
	if ( ldap_pvt_runqueue_isrunning( &slapd_rq, rtask )) {
		ldap_pvt_runqueue_stoptask( &slapd_rq, rtask );
	}
	/* If there were no queries, defer processing for a while */
	cm->cc_paused = pause;
	ldap_pvt_runqueue_resched( &slapd_rq, rtask, pause );

	ldap_pvt_thread_mutex_unlock( &slapd_rq.rq_mutex );
	return NULL;
}


#define MAX_ATTR_SETS 500

enum {
	PC_MAIN = 1,
	PC_ATTR,
	PC_TEMP,
	PC_RESP,
	PC_QUERIES
};

static ConfigDriver pc_cf_gen;
static ConfigLDAPadd pc_ldadd;
static ConfigCfAdd pc_cfadd;

static ConfigTable pccfg[] = {
	{ "proxycache", "backend> <max_entries> <numattrsets> <entry limit> "
				"<cycle_time",
		6, 6, 0, ARG_MAGIC|ARG_NO_DELETE|PC_MAIN, pc_cf_gen,
		"( OLcfgOvAt:2.1 NAME 'olcProxyCache' "
			"DESC 'ProxyCache basic parameters' "
			"SYNTAX OMsDirectoryString SINGLE-VALUE )", NULL, NULL },
	{ "proxyattrset", "index> <attributes...",
		2, 0, 0, ARG_MAGIC|PC_ATTR, pc_cf_gen,
		"( OLcfgOvAt:2.2 NAME 'olcProxyAttrset' "
			"DESC 'A set of attributes to cache' "
			"SYNTAX OMsDirectoryString )", NULL, NULL },
	{ "proxytemplate", "filter> <attrset-index> <TTL> <negTTL",
		4, 5, 0, ARG_MAGIC|PC_TEMP, pc_cf_gen,
		"( OLcfgOvAt:2.3 NAME 'olcProxyTemplate' "
			"DESC 'Filter template, attrset, cache TTL, optional negative TTL' "
			"SYNTAX OMsDirectoryString )", NULL, NULL },
	{ "response-callback", "head|tail(default)",
		2, 2, 0, ARG_MAGIC|PC_RESP, pc_cf_gen,
		"( OLcfgOvAt:2.4 NAME 'olcProxyResponseCB' "
			"DESC 'Response callback position in overlay stack' "
			"SYNTAX OMsDirectoryString )", NULL, NULL },
	{ "proxyCacheQueries", "queries",
		2, 2, 0, ARG_INT|ARG_MAGIC|PC_QUERIES, pc_cf_gen,
		"( OLcfgOvAt:2.5 NAME 'olcProxyCacheQueries' "
			"DESC 'Maximum number of queries to cache' "
			"SYNTAX OMsInteger )", NULL, NULL },

	{ NULL, NULL, 0, 0, 0, ARG_IGNORED }
};

static ConfigOCs pcocs[] = {
	{ "( OLcfgOvOc:2.1 "
		"NAME 'olcPcacheConfig' "
		"DESC 'ProxyCache configuration' "
		"SUP olcOverlayConfig "
		"MUST ( olcProxyCache $ olcProxyAttrset $ olcProxyTemplate ) "
		"MAY ( olcProxyResponseCB $ olcProxyCacheQueries ) )", Cft_Overlay, pccfg, NULL, pc_cfadd },
	{ "( OLcfgOvOc:2.2 "
		"NAME 'olcPcacheDatabase' "
		"DESC 'Cache database configuration' "
		"AUXILIARY )", Cft_Misc, pccfg, pc_ldadd },
	{ NULL, 0, NULL }
};

static int
pc_ldadd( CfEntryInfo *p, Entry *e, ConfigArgs *ca )
{
	slap_overinst *on;
	cache_manager *cm;

	if ( p->ce_type != Cft_Overlay || !p->ce_bi ||
		p->ce_bi->bi_cf_ocs != pcocs )
		return LDAP_CONSTRAINT_VIOLATION;

	on = (slap_overinst *)p->ce_bi;
	cm = on->on_bi.bi_private;
	ca->be = &cm->db;
	return LDAP_SUCCESS;
}

static int
pc_cfadd( Operation *op, SlapReply *rs, Entry *p, ConfigArgs *ca )
{
	CfEntryInfo *pe = p->e_private;
	slap_overinst *on = (slap_overinst *)pe->ce_bi;
	cache_manager *cm = on->on_bi.bi_private;
	struct berval bv;

	/* FIXME: should not hardcode "olcDatabase" here */
	bv.bv_len = sprintf( ca->msg, "olcDatabase=%s", cm->db.bd_info->bi_type );
	bv.bv_val = ca->msg;
	ca->be = &cm->db;

	/* We can only create this entry if the database is table-driven
	 */
	if ( cm->db.bd_info->bi_cf_ocs )
		config_build_entry( op, rs, pe, ca, &bv, cm->db.bd_info->bi_cf_ocs,
			&pcocs[1] );

	return 0;
}

static int
pc_cf_gen( ConfigArgs *c )
{
	slap_overinst	*on = (slap_overinst *)c->bi;
	cache_manager* 	cm = on->on_bi.bi_private;
	query_manager*  qm = cm->qm;
	QueryTemplate* 	temp;
	AttributeName*  attr_name;
	AttributeName* 	attrarray;
	const char* 	text=NULL;
	int		i, num, rc = 0;
	char		*ptr;
	unsigned long	t;

	if ( c->op == SLAP_CONFIG_EMIT ) {
		struct berval bv;
		switch( c->type ) {
		case PC_MAIN:
			bv.bv_len = snprintf( c->msg, sizeof( c->msg ), "%s %d %d %d %ld",
				cm->db.bd_info->bi_type, cm->max_entries, cm->numattrsets,
				cm->num_entries_limit, cm->cc_period );
			bv.bv_val = c->msg;
			value_add_one( &c->rvalue_vals, &bv );
			break;
		case PC_ATTR:
			for (i=0; i<cm->numattrsets; i++) {
				if ( !qm->attr_sets[i].count ) continue;

				bv.bv_len = snprintf( c->msg, sizeof( c->msg ), "%d", i );

				/* count the attr length */
				for ( attr_name = qm->attr_sets[i].attrs;