Reschedule pending netreqs

src/context.c

@@ -1267,6 +1267,26 @@ NULL_update_callback(
     getdns_context *context, getdns_context_code_t code, void *userarg)
 { (void)context; (void)code; (void)userarg; }
 
+static int
+netreq_expiry_cmp(const void *id1, const void *id2)
+{
+	getdns_network_req *req1 = (getdns_network_req *)id1;
+	getdns_network_req *req2 = (getdns_network_req *)id2;
+
+	return req1->owner->expires < req2->owner->expires ? -1 :
+	       req1->owner->expires > req2->owner->expires ?  1 :
+	       req1 < req2 ? -1 :
+	       req1 > req2 ?  1 : 0;
+}
+
+void _getdns_check_expired_pending_netreqs(
+    getdns_context *context, uint64_t *now_ms);
+static void _getdns_check_expired_pending_netreqs_cb(void *arg)
+{
+	uint64_t now_ms = 0;
+	_getdns_check_expired_pending_netreqs((getdns_context *)arg, &now_ms);
+}
+
 /*
  * getdns_context_create
  *
@@ -1330,8 +1350,15 @@ getdns_context_create_with_extended_memory_functions(
 
 	_getdns_rbtree_init(&result->outbound_requests, transaction_id_cmp);
 	_getdns_rbtree_init(&result->local_hosts, local_host_cmp);
-	/* TODO: Initialize pending_netreqs */
+	_getdns_rbtree_init(&result->pending_netreqs, netreq_expiry_cmp);
+	result->first_pending_netreq = NULL;
 	result->netreqs_in_flight = 0;
+	result->pending_timeout_event.userarg    = result;
+	result->pending_timeout_event.read_cb    = NULL;
+	result->pending_timeout_event.write_cb   = NULL;
+	result->pending_timeout_event.timeout_cb =
+	    _getdns_check_expired_pending_netreqs_cb;
+	result->pending_timeout_event.ev         = NULL;
 
 	result->server = NULL;
 
@@ -2982,27 +3009,6 @@ _getdns_context_request_timed_out(getdns_dns_req *dnsreq)
 	_getdns_context_cancel_request(dnsreq);
 }
 
-
-void
-_getdns_netreq_change_state(getdns_network_req *netreq, network_req_state new_state)
-{
-	if (!netreq)
-		return;
-
-	if (netreq->state != NET_REQ_IN_FLIGHT) {
-		if (new_state == NET_REQ_IN_FLIGHT)
-			netreq->owner->context->netreqs_in_flight += 1;
-		netreq->state = new_state;
-		return;
-	}
-	if (new_state == NET_REQ_IN_FLIGHT) /* No change */
-		return;
-	netreq->state = new_state;
-	netreq->owner->context->netreqs_in_flight -= 1;
-	/* TODO: Schedule pending netreqs
-	 *       when netreqs_in_flight < oustanding_queries */
-}
-
 static void
 accumulate_outstanding_transactions(_getdns_rbnode_t *node, void* arg)
 {

src/context.h

@@ -294,7 +294,10 @@ struct getdns_context {
 	/* network requests
 	 */
 	size_t netreqs_in_flight;
-	_getdns_rbtree_t pending_netreqs;
+
+	_getdns_rbtree_t       pending_netreqs;
+	getdns_network_req    *first_pending_netreq;
+	getdns_eventloop_event pending_timeout_event;
 
 	struct listen_set *server;
 
@@ -382,17 +385,6 @@ void _getdns_context_cancel_request(getdns_dns_req *dnsreq);
  */
 void _getdns_context_request_timed_out(getdns_dns_req *dnsreq);
 
-/* Change state of the netreq req.
- * - Increments context->netreqs_in_flight
- *   when state changes from NOT_SENT to IN_FLIGHT
- * - Decrements context->netreqs_in_flight
- *   when state changes from IN_FLIGHT to FINISHED, TIMED_OUT or ERRORED
- * - Resubmits NOT_SENT netreqs from context->pending_netreqs,
- *   when # pending_netreqs < limit_outstanding_queries
- */
-void _getdns_netreq_change_state(
-    getdns_network_req *req, network_req_state new_state);
-
 char *_getdns_strdup(const struct mem_funcs *mfs, const char *str);
 
 struct getdns_bindata *_getdns_bindata_copy(

src/general.c

@@ -90,14 +90,23 @@ void
 _getdns_check_dns_req_complete(getdns_dns_req *dns_req)
 {
 	getdns_network_req **netreq_p, *netreq;
-	int results_found = 0, r;
+	int results_found = 0, timed_out = 1, r;
 	uint64_t now_ms  = 0;
 	
 	for (netreq_p = dns_req->netreqs; (netreq = *netreq_p); netreq_p++)
 		if (!_getdns_netreq_finished(netreq))
 			return;
-		else if (netreq->response_len > 0)
+		else {
-			results_found = 1;
+			if (netreq->state != NET_REQ_TIMED_OUT)
+				timed_out = 0;
+			if (netreq->response_len > 0)
+				results_found = 1;
+		}
+
+	if (timed_out) {
+		_getdns_context_request_timed_out(dns_req);
+		return;
+	}
 
 	/* Do we have to check more suffixes on nxdomain/nodata?
 	 */
@@ -248,30 +257,113 @@ ub_resolve_callback(void* arg, int err, struct ub_result* ub_res)
 #endif
 
 
+void _getdns_check_expired_pending_netreqs(
+    getdns_context *context, uint64_t *now_ms)
+{
+	getdns_network_req *first;
+
+	assert(context);
+
+	while (context->pending_netreqs.count) {
+		first = (getdns_network_req *)
+		    _getdns_rbtree_first(&context->pending_netreqs);
+
+		if (_getdns_ms_until_expiry2(first->owner->expires, now_ms) > 0)
+			break;
+
+		(void) _getdns_rbtree_delete(&context->pending_netreqs, first);
+		_getdns_netreq_change_state(first, NET_REQ_TIMED_OUT);
+		_getdns_check_dns_req_complete(first->owner);
+	}
+	first = context->pending_netreqs.count ? (getdns_network_req *)
+	    _getdns_rbtree_first(&context->pending_netreqs) : NULL;
+
+	if (first == context->first_pending_netreq ||
+	    (first && context->first_pending_netreq &&
+	     first->owner->expires == context->first_pending_netreq->owner->expires))
+		return; /* Nothing changed */
+
+	if (context->first_pending_netreq)
+		GETDNS_CLEAR_EVENT(  context->extension
+		                  , &context->pending_timeout_event);
+
+	if ((context->first_pending_netreq = first))
+		GETDNS_SCHEDULE_EVENT( context->extension, -1,
+		    _getdns_ms_until_expiry2(first->owner->expires, now_ms),
+		    &context->pending_timeout_event);
+}
+
+void
+_getdns_netreq_change_state(
+    getdns_network_req *netreq, network_req_state new_state)
+{
+	getdns_context *context;
+	uint64_t now_ms;
+
+	if (!netreq)
+		return;
+
+	context = netreq->owner->context;
+
+	if (netreq->state != NET_REQ_IN_FLIGHT) {
+		if (new_state == NET_REQ_IN_FLIGHT)
+			context->netreqs_in_flight += 1;
+		netreq->state = new_state;
+		return;
+	}
+	if (new_state == NET_REQ_IN_FLIGHT) /* No change */
+		return;
+	netreq->state = new_state;
+	context->netreqs_in_flight -= 1;
+
+	now_ms = 0;
+	while (context->limit_outstanding_queries > 0 &&
+	    context->pending_netreqs.count > 0 &&
+	    context->netreqs_in_flight < context->limit_outstanding_queries)  {
+
+		getdns_network_req *first = (getdns_network_req *)
+		    _getdns_rbtree_first(&context->pending_netreqs);
+		(void) _getdns_rbtree_delete(&context->pending_netreqs, first);
+		(void) _getdns_submit_netreq(first, &now_ms);
+	}
+}
+
 int
 _getdns_submit_netreq(getdns_network_req *netreq, uint64_t *now_ms)
 {
 	getdns_return_t r;
 	getdns_dns_req *dns_req = netreq->owner;
+	getdns_context *context = dns_req->context;
 	char name[1024];
 	int dnsreq_freed = 0;
 #ifdef HAVE_LIBUNBOUND
 	int ub_resolve_r;
 #endif
 
+	if (context->limit_outstanding_queries > 0 &&
+	    context->netreqs_in_flight >= context->limit_outstanding_queries) {
+
+		netreq->node.key = netreq;
+		if (_getdns_rbtree_insert(
+		    &context->pending_netreqs, &netreq->node)) {
+			
+			_getdns_check_expired_pending_netreqs(context, now_ms);
+			return GETDNS_RETURN_GOOD;
+		}
+	}
 	_getdns_netreq_change_state(netreq, NET_REQ_IN_FLIGHT);
 
 #ifdef STUB_NATIVE_DNSSEC
 # ifdef DNSSEC_ROADBLOCK_AVOIDANCE
 
-	if ((dns_req->context->resolution_type == GETDNS_RESOLUTION_RECURSING
+	if ((context->resolution_type == GETDNS_RESOLUTION_RECURSING
 	    && !dns_req->dnssec_roadblock_avoidance)
 	    ||  dns_req->avoid_dnssec_roadblocks) {
 # else
-	if ( dns_req->context->resolution_type == GETDNS_RESOLUTION_RECURSING) {
+	if ( context->resolution_type == GETDNS_RESOLUTION_RECURSING) {
 # endif
 #else
-	if ( dns_req->context->resolution_type == GETDNS_RESOLUTION_RECURSING
+	if ( context->resolution_type == GETDNS_RESOLUTION_RECURSING
 	    || dns_req->dnssec_return_status
 	    || dns_req->dnssec_return_only_secure
 	    || dns_req->dnssec_return_all_statuses
@@ -297,15 +389,15 @@ _getdns_submit_netreq(getdns_network_req *netreq, uint64_t *now_ms)
 #ifdef HAVE_LIBUNBOUND
 		dns_req->freed = &dnsreq_freed;
 #ifdef HAVE_UNBOUND_EVENT_API
-		if (_getdns_ub_loop_enabled(&dns_req->context->ub_loop))
+		if (_getdns_ub_loop_enabled(&context->ub_loop))
-			ub_resolve_r = ub_resolve_event(dns_req->context->unbound_ctx,
+			ub_resolve_r = ub_resolve_event(context->unbound_ctx,
-			    name, netreq->request_type, netreq->owner->request_class,
+			    name, netreq->request_type, dns_req->request_class,
 			    netreq, ub_resolve_event_callback, &(netreq->unbound_id)) ?
 			    GETDNS_RETURN_GENERIC_ERROR : GETDNS_RETURN_GOOD;
 		else
 #endif
-			ub_resolve_r = ub_resolve_async(dns_req->context->unbound_ctx,
+			ub_resolve_r = ub_resolve_async(context->unbound_ctx,
-			    name, netreq->request_type, netreq->owner->request_class,
+			    name, netreq->request_type, dns_req->request_class,
 			    netreq, ub_resolve_callback, &(netreq->unbound_id)) ?
 			    GETDNS_RETURN_GENERIC_ERROR : GETDNS_RETURN_GOOD;
 		if (dnsreq_freed)

src/general.h

@@ -45,6 +45,18 @@
 #define DNS_REQ_FINISHED -1
 
 void _getdns_call_user_callback(getdns_dns_req *, getdns_dict *);
+
+/* Change state of the netreq req.
+ * - Increments context->netreqs_in_flight
+ *   when state changes from NOT_SENT to IN_FLIGHT
+ * - Decrements context->netreqs_in_flight
+ *   when state changes from IN_FLIGHT to FINISHED, TIMED_OUT or ERRORED
+ * - Resubmits NOT_SENT netreqs from context->pending_netreqs,
+ *   when # pending_netreqs < limit_outstanding_queries
+ */
+void _getdns_netreq_change_state(
+    getdns_network_req *netreq, network_req_state new_state);
+
 void _getdns_check_dns_req_complete(getdns_dns_req *dns_req);
 int _getdns_submit_netreq(getdns_network_req *netreq, uint64_t *now_ms);
 

src/request-internal.c

@@ -43,6 +43,7 @@
 #include "dict.h"
 #include "debug.h"
 #include "convert.h"
+#include "general.h"
 
 /* MAXIMUM_TSIG_SPACE = TSIG name      (dname)    : 256
  *                      TSIG type      (uint16_t) :   2

src/test/tpkg/280-limit_outstanding_queries.tpkg/280-limit_outstanding_queries.test

@@ -4,9 +4,15 @@
 # use .tpkg.var.test for in test variable passing
 [ -f .tpkg.var.test ] && source .tpkg.var.test
 
-# TODO: Change QLIMIT to 10 once implement pending netreq resubmission
+
-#
+QLIMIT=64
-QLIMIT=1000
+NQUERIES=`wc "./${TPKG_NAME}.queries"|sed 's/ .*$//g'`
+
+# Test will take NQUERIES / QLIMIT * answer delay
+# For current parameters this is 1000 / 64 * 0.3 = 4.6875
+# which is smaller than 5 seconds default query timeout value,
+# so the test should succeed.
+
 make && "./${TPKG_NAME}" | (
 	read PORT
 	${GETDNS_STUB_QUERY} @127.0.0.1:$PORT TXT \
@@ -15,7 +21,7 @@ make && "./${TPKG_NAME}" | (
 
 	${GETDNS_STUB_QUERY} -q @127.0.0.1:$PORT TXT quit.
 ) && grep '"n_requests: [0-9][0-9]*"' out | sed -e 's/^.*n_requests: //g' -e 's/".*$//g' \
-    | awk -vQLIMIT=$QLIMIT '
+    | awk -vQLIMIT=$QLIMIT -vNQUERIES=$NQUERIES '
 
 BEGIN{
 	max_outstanding = 0;
@@ -25,6 +31,7 @@ BEGIN{
 		max_outstanding = $1;
 }
 END{
+	printf("%d of %d queries answered (%.1f%%)\n", NR, NQUERIES, (NR / NQUERIES * 100));
 	if (max_outstanding > QLIMIT) {
 		print "ERROR: More than "QLIMIT" outstanding queries: "max_outstanding;
 		exit(-1);