Add prefetch support for subnet cache entries

- Entries in the subnet cache should now be prefetched.

- Rename testdata subnet_*.crpl to subnet_*.rpl so they are visible to
  make test

Signed-off-by: Tian Lan <tian.lan@twosigma.com>

edns-subnet/subnetmod.c

@@ -97,8 +97,8 @@ subnet_new_qstate(struct module_qstate *qstate, int id)
 }
 
 /** Add ecs struct to edns list, after parsing it to wire format. */
-static void
-ecs_opt_list_append(struct ecs_data* ecs, struct edns_option** list,
+void
+subnet_ecs_opt_list_append(struct ecs_data* ecs, struct edns_option** list,
 	struct module_qstate *qstate)
 {
 	size_t sn_octs, sn_octs_remainder;
@@ -164,7 +164,7 @@ int ecs_whitelist_check(struct query_info* qinfo,
 		 * set. */
 		if(!edns_opt_list_find(qstate->edns_opts_back_out,
 			qstate->env->cfg->client_subnet_opcode)) {
-			ecs_opt_list_append(&sq->ecs_server_out,
+			subnet_ecs_opt_list_append(&sq->ecs_server_out,
 				&qstate->edns_opts_back_out, qstate);
 		}
 		sq->subnet_sent = 1;
@@ -231,7 +231,7 @@ subnetmod_init(struct module_env *env, int id)
 	env->unique_mesh = 1;
 	if(!edns_register_option(env->cfg->client_subnet_opcode,
 		env->cfg->client_subnet_always_forward /* bypass cache */,
-		0 /* no aggregation */, env)) {
+		1 /* no aggregation */, env)) {
 		log_err("subnetcache: could not register opcode");
 		ecs_whitelist_delete(sn_env->whitelist);
 		slabhash_delete(sn_env->subnet_msg_cache);
@@ -330,11 +330,15 @@ update_cache(struct module_qstate *qstate, int id)
 	struct slabhash *subnet_msg_cache = sne->subnet_msg_cache;
 	struct ecs_data *edns = &sq->ecs_client_in;
 	size_t i;
+	hashvalue_type h;
+
+	/* qinfo_hash is not set if it is prefetch request */
+	if (qstate->minfo[id] && ((struct subnet_qstate*)qstate->minfo[id])->qinfo_hash) {
+		h = ((struct subnet_qstate*)qstate->minfo[id])->qinfo_hash;
+	} else {
+		h = query_info_hash(&qstate->qinfo, qstate->query_flags);
+	}
 
-	/* We already calculated hash upon lookup */
-	hashvalue_type h = qstate->minfo[id] ? 
-		((struct subnet_qstate*)qstate->minfo[id])->qinfo_hash : 
-		query_info_hash(&qstate->qinfo, qstate->query_flags);
 	/* Step 1, general qinfo lookup */
 	struct lruhash_entry *lru_entry = slabhash_lookup(subnet_msg_cache, h,
 		&qstate->qinfo, 1);
@@ -380,7 +384,7 @@ update_cache(struct module_qstate *qstate, int id)
 		log_err("subnetcache: cache insertion failed");
 		return;
 	}
-	
+
 	/* store RRsets */
 	for(i=0; i<rep->rrset_count; i++) {
 		rep->ref[i].key = rep->rrsets[i];
@@ -402,7 +406,7 @@ update_cache(struct module_qstate *qstate, int id)
 
 /** Lookup in cache and reply true iff reply is sent. */
 static int
-lookup_and_reply(struct module_qstate *qstate, int id, struct subnet_qstate *sq)
+lookup_and_reply(struct module_qstate *qstate, int id, struct subnet_qstate *sq, int prefetch)
 {
 	struct lruhash_entry *e;
 	struct module_env *env = qstate->env;
@@ -451,6 +455,10 @@ lookup_and_reply(struct module_qstate *qstate, int id, struct subnet_qstate *sq)
 			INET6_SIZE);
 		sq->ecs_client_out.subnet_validdata = 1;
 	}
+
+	if (prefetch && *qstate->env->now > ((struct reply_info *)node->elem)->prefetch_ttl) {
+		qstate->need_refetch = 1;
+	}
 	return 1;
 }
 
@@ -487,7 +495,7 @@ eval_response(struct module_qstate *qstate, int id, struct subnet_qstate *sq)
 		 * module_finished */
 		return module_finished;
 	}
-	
+
 	/* We have not asked for subnet data */
 	if (!sq->subnet_sent) {
 		if (s_in->subnet_validdata)
@@ -496,7 +504,7 @@ eval_response(struct module_qstate *qstate, int id, struct subnet_qstate *sq)
 			cp_edns_bad_response(c_out, c_in);
 		return module_finished;
 	}
-	
+
 	/* subnet sent but nothing came back */
 	if (!s_in->subnet_validdata) {
 		/* The authority indicated no support for edns subnet. As a
@@ -513,11 +521,11 @@ eval_response(struct module_qstate *qstate, int id, struct subnet_qstate *sq)
 			cp_edns_bad_response(c_out, c_in);
 		return module_finished;
 	}
-	
+
 	/* Being here means we have asked for and got a subnet specific 
 	 * answer. Also, the answer from the authority is not yet cached 
 	 * anywhere. */
-	
+
 	/* can we accept response? */
 	if(s_out->subnet_addr_fam != s_in->subnet_addr_fam ||
 		s_out->subnet_source_mask != s_in->subnet_source_mask ||
@@ -602,7 +610,7 @@ parse_subnet_option(struct edns_option* ecs_option, struct ecs_data* ecs)
 	return 1;
 }
 
-static void
+void
 subnet_option_from_ss(struct sockaddr_storage *ss, struct ecs_data* ecs,
 	struct config_file* cfg)
 {
@@ -759,13 +767,13 @@ subnetmod_operate(struct module_qstate *qstate, enum module_ev event,
 		}
 
 		lock_rw_wrlock(&sne->biglock);
-		if (lookup_and_reply(qstate, id, sq)) {
+		if (qstate->mesh_info->reply_list && lookup_and_reply(qstate, id, sq, qstate->env->cfg->prefetch)) {
 			sne->num_msg_cache++;
 			lock_rw_unlock(&sne->biglock);
 			verbose(VERB_QUERY, "subnetcache: answered from cache");
 			qstate->ext_state[id] = module_finished;
 
-			ecs_opt_list_append(&sq->ecs_client_out,
+			subnet_ecs_opt_list_append(&sq->ecs_client_out,
 				&qstate->edns_opts_front_out, qstate);
 			return;
 		}
@@ -787,7 +795,7 @@ subnetmod_operate(struct module_qstate *qstate, enum module_ev event,
 			sq->ecs_server_out.subnet_source_mask =
 				qstate->env->cfg->max_client_subnet_ipv6;
 		/* Safe to copy completely, even if the source is limited by the
-		 * configuration. ecs_opt_list_append() will limit the address.
+		 * configuration. subnet_ecs_opt_list_append() will limit the address.
 		 * */
 		memcpy(&sq->ecs_server_out.subnet_addr,
 			sq->ecs_client_in.subnet_addr, INET6_SIZE);
@@ -811,7 +819,7 @@ subnetmod_operate(struct module_qstate *qstate, enum module_ev event,
 		qstate->ext_state[id] = eval_response(qstate, id, sq);
 		if(qstate->ext_state[id] == module_finished &&
 			qstate->return_msg) {
-			ecs_opt_list_append(&sq->ecs_client_out,
+			subnet_ecs_opt_list_append(&sq->ecs_client_out,
 				&qstate->edns_opts_front_out, qstate);
 		}
 		qstate->no_cache_store = sq->started_no_cache_store;

edns-subnet/subnetmod.h

@@ -143,4 +143,12 @@ int ecs_query_response(struct module_qstate* qstate, struct dns_msg* response,
 /** mark subnet msg to be deleted */
 void subnet_markdel(void* key);
 
+/** Add ecs struct to edns list, after parsing it to wire format. */
+void subnet_ecs_opt_list_append(struct ecs_data* ecs, struct edns_option** list,
+	struct module_qstate *qstate);
+
+/** Create ecs_data from the sockaddr_storage information. */
+void subnet_option_from_ss(struct sockaddr_storage *ss, struct ecs_data* ecs,
+	struct config_file* cfg);
+
 #endif /* SUBNETMOD_H */

services/mesh.c

@@ -64,6 +64,11 @@
 #include "respip/respip.h"
 #include "services/listen_dnsport.h"
 
+#ifdef CLIENT_SUBNET
+#include "edns-subnet/subnetmod.h"
+#include "edns-subnet/edns-subnet.h"
+#endif
+
 /** subtract timers and the values do not overflow or become negative */
 static void
 timeval_subtract(struct timeval* d, const struct timeval* end, const struct timeval* start)
@@ -683,6 +688,107 @@ mesh_new_callback(struct mesh_area* mesh, struct query_info* qinfo,
 	return 1;
 }
 
+#ifdef CLIENT_SUBNET
+/* Same logic as mesh_schedule_prefetch but tailored to the subnet module logic
+ * like passing along the comm_reply info. This will be faked into an EDNS
+ * option for processing by the subnet module if the client has not already
+ * attached its own ECS data. */
+static void mesh_schedule_prefetch_subnet(struct mesh_area* mesh,
+	struct query_info* qinfo, uint16_t qflags, time_t leeway, int run,
+	int rpz_passthru, struct mesh_state* mstate,
+	struct sockaddr_storage *client_addr)
+{
+	struct mesh_state* s = NULL;
+	struct edns_option* opt = NULL;
+#ifdef UNBOUND_DEBUG
+	struct rbnode_type* n;
+#endif
+
+	if(!mesh_make_new_space(mesh, NULL)) {
+		verbose(VERB_ALGO, "Too many queries. dropped prefetch.");
+		mesh->stats_dropped ++;
+		return;
+	}
+
+	s = mesh_state_create(mesh->env, qinfo, NULL,
+		qflags&(BIT_RD|BIT_CD), 0, 0);
+	if(!s) {
+		log_err("prefetch_subnet mesh_state_create: out of memory");
+		return;
+	}
+
+	mesh_state_make_unique(s);
+
+	opt = edns_opt_list_find(mstate->s.edns_opts_front_in, mesh->env->cfg->client_subnet_opcode);
+	if(opt) {
+		/* Use the client's ECS data */
+		if(!edns_opt_list_append(&s->s.edns_opts_front_in, opt->opt_code,
+			opt->opt_len, opt->opt_data, s->s.region)) {
+			log_err("prefetch_subnet edns_opt_list_append: out of memory");
+			return;
+		}
+	} else {
+		/* Fake the ECS data from the client's IP */
+		struct ecs_data ecs;
+		memset(&ecs, 0, sizeof(ecs));
+		subnet_option_from_ss(client_addr, &ecs, mesh->env->cfg);
+
+		if(ecs.subnet_validdata == 0) {
+			log_err("prefetch_subnet subnet_option_from_ss: invalid data");
+			return;
+		}
+
+		subnet_ecs_opt_list_append(&ecs, &s->s.edns_opts_front_in, &s->s);
+		if(!s->s.edns_opts_front_in) {
+			log_err("prefetch_subnet subnet_ecs_opt_list_append: out of memory");
+			return;
+		}
+	}
+#ifdef UNBOUND_DEBUG
+	n =
+#else
+	(void)
+#endif
+	rbtree_insert(&mesh->all, &s->node);
+	log_assert(n != NULL);
+	/* set detached (it is now) */
+	mesh->num_detached_states++;
+	/* make it ignore the cache */
+	sock_list_insert(&s->s.blacklist, NULL, 0, s->s.region);
+	s->s.prefetch_leeway = leeway;
+
+	if(s->list_select == mesh_no_list) {
+		/* move to either the forever or the jostle_list */
+		if(mesh->num_forever_states < mesh->max_forever_states) {
+			mesh->num_forever_states ++;
+			mesh_list_insert(s, &mesh->forever_first,
+				&mesh->forever_last);
+			s->list_select = mesh_forever_list;
+		} else {
+			mesh_list_insert(s, &mesh->jostle_first,
+				&mesh->jostle_last);
+			s->list_select = mesh_jostle_list;
+		}
+	}
+
+	s->s.rpz_passthru = rpz_passthru;
+
+	if(!run) {
+#ifdef UNBOUND_DEBUG
+		n =
+#else
+		(void)
+#endif
+		rbtree_insert(&mesh->run, &s->run_node);
+		log_assert(n != NULL);
+		return;
+	}
+
+	mesh_state_delete(&mstate->s);
+	mesh_run(mesh, s, module_event_new, NULL);
+}
+#endif /* CLIENT_SUBNET */
+
 /* Internal backend routine of mesh_new_prefetch().  It takes one additional
  * parameter, 'run', which controls whether to run the prefetch state
  * immediately.  When this function is called internally 'run' could be
@@ -1699,6 +1805,11 @@ mesh_continue(struct mesh_area* mesh, struct mesh_state* mstate,
 			struct query_info* qinfo = NULL;
 			uint16_t qflags;
 			int rpz_p = 0;
+			struct sockaddr_storage client_addr;
+
+			if (mstate->reply_list) {
+				client_addr = mstate->reply_list->query_reply.addr;
+			}
 
 			mesh_query_done(mstate);
 			mesh_walk_supers(mesh, mstate);
@@ -1712,10 +1823,24 @@ mesh_continue(struct mesh_area* mesh, struct mesh_state* mstate,
 				rpz_p = mstate->s.rpz_passthru;
 			}
 
-			mesh_state_delete(&mstate->s);
 			if(qinfo) {
+#ifdef CLIENT_SUBNET
+				if(modstack_find(&mesh->mods, "subnetcache") != -1 ) {
+					mesh_schedule_prefetch_subnet(mesh, qinfo, qflags,
+						0, 1, rpz_p, mstate, &client_addr);
+				}
+				else {
+					mesh_state_delete(&mstate->s);
+					mesh_schedule_prefetch(mesh, qinfo, qflags,
+						0, 1, rpz_p);
+				}
+#else
+				mesh_state_delete(&mstate->s);
 				mesh_schedule_prefetch(mesh, qinfo, qflags,
 					0, 1, rpz_p);
+#endif
+			} else {
+				mesh_state_delete(&mstate->s);
 			}
 			return 0;
 		}