545cdd4b50

lib/asyn-thrdd.c+142 -35

@@ -293,8 +293,12 @@ void Curl_async_thrdd_destroy(struct Curl_easy *data,
	curlx_safefree(async->thrdd.rr.https_name);
	Curl_httpsrr_destroy(async->thrdd.rr.hinfo);
	#endif
+	async_thrdd_item_destroy(async->thrdd.inc_A);
+	async->thrdd.inc_A = NULL;
	async_thrdd_item_destroy(async->thrdd.res_A);
	async->thrdd.res_A = NULL;
+	async_thrdd_item_destroy(async->thrdd.inc_AAAA);
+	async->thrdd.inc_AAAA = NULL;
	async_thrdd_item_destroy(async->thrdd.res_AAAA);
	async->thrdd.res_AAAA = NULL;
	}
@@ -309,28 +313,32 @@ CURLcode Curl_async_await(struct Curl_easy *data, uint32_t resolv_id,
	struct Curl_resolv_async *async = Curl_async_get(data, resolv_id);
	struct async_thrdd_ctx *thrdd = async ? &async->thrdd : NULL;
	timediff_t milli, ms;
+	CURLcode result = CURLE_AGAIN;

	if(!thrdd)
	return CURLE_FAILED_INIT;

-	while(async->queries_ongoing && !async->done) {
-	Curl_async_thrdd_multi_process(data->multi);
-	if(async->done)
-	break;
-
-	ms = curlx_ptimediff_ms(Curl_pgrs_now(data), &async->start);
-	if(ms < 3)
-	milli = 0;
-	else if(ms <= 50)
-	milli = ms / 3;
-	else if(ms <= 250)
-	milli = 50;
-	else
-	milli = 200;
-	CURL_TRC_DNS(data, "await, waiting %" FMT_TIMEDIFF_T "ms", milli);
-	curlx_wait_ms(milli);
+	while(result == CURLE_AGAIN) {
+	while(async->queries_ongoing && !async->done) {
+	Curl_async_thrdd_multi_process(data->multi);
+	if(async->done)
+	break;
+
+	ms = curlx_ptimediff_ms(Curl_pgrs_now(data), &async->start);
+	if(ms < 3)
+	milli = 0;
+	else if(ms <= 50)
+	milli = ms / 3;
+	else if(ms <= 250)
+	milli = 50;
+	else
+	milli = 200;
+	CURL_TRC_DNS(data, "await, waiting %" FMT_TIMEDIFF_T "ms", milli);
+	curlx_wait_ms(milli);
+	}
+	result = Curl_async_take_result(data, async, pdns);
	}
-	return Curl_async_take_result(data, async, pdns);
+	return result;
	}

	#ifdef HAVE_GETADDRINFO
@@ -469,7 +477,7 @@ CURLcode Curl_async_thrdd_multi_init(struct Curl_multi *multi,
	{
	CURLcode result;
	DEBUGASSERT(!multi->resolv_thrdq);
-	result = Curl_thrdq_create(&multi->resolv_thrdq, "DNS", 0,
+	result = Curl_thrdq_create(&multi->resolv_thrdq, "DNS",
	min_threads, max_threads, idle_time_ms,
	async_thrdd_item_free,
	async_thrdd_item_process,
@@ -561,30 +569,25 @@ void Curl_async_thrdd_multi_process(struct Curl_multi *multi)
	if(data)
	async = Curl_async_get(data, item->resolv_id);
	if(async) {
-	struct async_thrdd_item **pdest = &async->thrdd.res_A;
-
-	async->dns_responses \|= item->dns_queries;
-	--async->queries_ongoing;
-	async->done = !async->queries_ongoing;
+	struct async_thrdd_item **pdest = &async->thrdd.inc_A;

	#ifdef CURLRES_IPV6
	if(item->dns_queries & CURL_DNSQ_AAAA)
-	pdest = &async->thrdd.res_AAAA;
+	pdest = &async->thrdd.inc_AAAA;
	#endif
	if(!*pdest) {
-	VERBOSE(async_thrdd_report_item(data, item));
	*pdest = item;
	item = NULL;
	}
	else
	DEBUGASSERT(0); /* should not receive duplicates here */
+
+	--async->queries_ongoing;
	Curl_multi_mark_dirty(data);
	}
	async_thrdd_item_free(item);
	}
-	#ifdef CURLVERBOSE
-	Curl_thrdq_trace(multi->resolv_thrdq, multi->admin);
-	#endif
+	VERBOSE(Curl_thrdq_trace(multi->resolv_thrdq, multi->admin));
	}

	CURLcode Curl_async_thrdd_multi_set_props(struct Curl_multi *multi,
@@ -592,7 +595,7 @@ CURLcode Curl_async_thrdd_multi_set_props(struct Curl_multi *multi,
	uint32_t max_threads,
	uint32_t idle_time_ms)
	{
-	return Curl_thrdq_set_props(multi->resolv_thrdq, 0,
+	return Curl_thrdq_set_props(multi->resolv_thrdq,
	min_threads, max_threads, idle_time_ms);
	}

@@ -723,6 +726,113 @@ CURLcode Curl_async_pollset(struct Curl_easy *data,
	return CURLE_OK;
	}

+	#if defined(USE_IPV6) && defined(HAVE_SOCKADDR_IN6_SIN6_SCOPE_ID)
+	static bool async_thrdd_item_missing_scope(struct Curl_easy *data,
+	struct async_thrdd_item *item)
+	{
+	const struct Curl_addrinfo *ai;
+
+	/* scope id already globally set */
+	if(data->conn && data->conn->scope_id)
+	return FALSE;
+
+	for(ai = item->res; ai; ai = ai->ai_next) {
+	if(ai->ai_family == AF_INET6) {
+	struct sockaddr_in6 sa6 = (void )ai->ai_addr;
+	if(IN6_IS_ADDR_LINKLOCAL(&sa6->sin6_addr) && !sa6->sin6_scope_id)
+	return TRUE;
+	}
+	}
+	return FALSE;
+	}
+
+	static void async_thrdd_item_strip_results(struct async_thrdd_item *item,
+	int ai_family)
+	{
+	struct Curl_addrinfo ai = item->res, *panchor = &item->res;
+	while(ai) {
+	if(ai->ai_family == ai_family) {
+	*panchor = ai->ai_next;
+	ai->ai_next = NULL;
+	Curl_freeaddrinfo(ai);
+	ai = *panchor;
+	}
+	else {
+	panchor = &ai->ai_next;
+	ai = ai->ai_next;
+	}
+	}
+	}
+
+	#endif /* USE_IPV6 && HAVE_SOCKADDR_IN6_SIN6_SCOPE_ID */
+
+	static CURLcode async_thrdd_check_done(struct Curl_easy *data,
+	struct Curl_resolv_async *async)
+	{
+	struct async_thrdd_ctx *thrdd = &async->thrdd;
+
+	(void)data;
+	if(thrdd->inc_A) {
+	thrdd->res_A = thrdd->inc_A;
+	thrdd->inc_A = NULL;
+	VERBOSE(async_thrdd_report_item(data, thrdd->res_A));
+	async->dns_responses \|= thrdd->res_A->dns_queries;
+	}
+
+	if(thrdd->inc_AAAA) {
+	if(thrdd->res_AAAA) {
+	DEBUGASSERT(0); /* should not happen */
+	return CURLE_FAILED_INIT;
+	}
+	#if defined(USE_IPV6) && defined(HAVE_SOCKADDR_IN6_SIN6_SCOPE_ID)
+	/* do we accept the incoming AAAA response? */
+	if(!(thrdd->inc_AAAA->dns_queries & CURL_DNSQ_A) &&
+	async_thrdd_item_missing_scope(data, thrdd->inc_AAAA)) {
+	/* We queried "only" AF_INET6. This may be problematic when the
+	* result has ipv6 link-local addresses and did not give
+	* any scope id for it. glibc has a long outstanding bug
+	* <https://sourceware.org/bugzilla/show_bug.cgi?id=14413>
+	* that gives scope ids only on AF_UNSPEC queries. */
+	struct async_thrdd_item *item = thrdd->inc_AAAA;
+	CURLcode result;
+
+	/* Reuse the item and queue it again, this time with added
+	* CURL_DNSQ_A which resolves using AF_UNSPEC. */
+	thrdd->inc_AAAA = NULL;
+	item->dns_queries \|= CURL_DNSQ_A;
+	if(item->res) {
+	Curl_freeaddrinfo(item->res);
+	item->res = NULL;
+	}
+
+	CURL_TRC_DNS(data, "re-queueing query %s for AF_UNSPEC resolve",
+	item->description);
+	result = Curl_thrdq_send(data->multi->resolv_thrdq, item,
+	async_item_description(item),
+	async->timeout_ms);
+	if(result) {
+	async_thrdd_item_free(item);
+	return result;
+	}
+	async->queries_ongoing++;
+	return CURLE_AGAIN;
+	}
+	/* accepting the AAAA result, strip it of any AF_INET entries,
+	* as we might have resolved it with AF_UNSPEC. */
+	async_thrdd_item_strip_results(thrdd->inc_AAAA, AF_INET);
+	#endif /* USE_IPV6 && HAVE_SOCKADDR_IN6_SIN6_SCOPE_ID */
+	thrdd->res_AAAA = thrdd->inc_AAAA;
+	thrdd->inc_AAAA = NULL;
+	VERBOSE(async_thrdd_report_item(data, thrdd->res_AAAA));
+	async->dns_responses \|= thrdd->res_AAAA->dns_queries;
+	}
+
+	if(async->queries_ongoing)
+	return CURLE_AGAIN;
+	async->done = TRUE;
+	return CURLE_OK;
+	}
+
	/*
	* Curl_async_take_result() is called repeatedly to check if a previous
	* name resolve request has completed. It should also make sure to time-out if
@@ -738,10 +848,6 @@ CURLcode Curl_async_take_result(struct Curl_easy *data,

	DEBUGASSERT(pdns);
	*pdns = NULL;
-	if(!async->queries_ongoing && !async->done) {
-	DEBUGASSERT(0);
-	return CURLE_FAILED_INIT;
-	}

	#ifdef USE_HTTPSRR_ARES
	/* best effort, ignore errors */
@@ -752,8 +858,9 @@ CURLcode Curl_async_take_result(struct Curl_easy *data,
	Curl_async_thrdd_multi_process(data->multi);
	#endif

-	if(!async->done)
-	return CURLE_AGAIN;
+	result = async_thrdd_check_done(data, async);
+	if(result)
+	return result;

	Curl_expire_done(data, EXPIRE_ASYNC_NAME);

lib/asyn.h+4 -2

@@ -140,8 +140,10 @@ struct async_thrdd_item;

	/* Context for threaded resolver */
	struct async_thrdd_ctx {
-	struct async_thrdd_item res_A; / IPv4 result */
-	struct async_thrdd_item res_AAAA; / IPv6 result */
+	struct async_thrdd_item inc_A; / IPv4 incoming result */
+	struct async_thrdd_item res_A; / IPv4 final result */
+	struct async_thrdd_item inc_AAAA; / IPv6 incoming result */
+	struct async_thrdd_item res_AAAA; / IPv6 final result */
	#if defined(USE_HTTPSRR) && defined(USE_ARES)
	struct {
	ares_channel channel;

lib/cf-dns.c+1 -1

@@ -291,7 +291,7 @@ static CURLcode cf_dns_connect(struct Curl_cfilter *cf,

	if(ctx->dns && !ctx->announced) {
	ctx->announced = TRUE;
-	if(cf->sockindex == FIRSTSOCKET) {
+	if((cf->sockindex == FIRSTSOCKET) && ip_query) {
	cf->conn->bits.dns_resolved = TRUE;
	Curl_pgrsTime(data, TIMER_NAMELOOKUP);
	}

lib/thrdqueue.c+20 -23

@@ -47,7 +47,6 @@ struct curl_thrdq {
	Curl_thrdq_item_process_cb *fn_process;
	Curl_thrdq_ev_cb *fn_event;
	void *fn_user_data;
-	uint32_t send_max_len;
	BIT(aborted);
	};

@@ -198,7 +197,6 @@ static void thrdq_unlink(struct curl_thrdq *tqueue, bool locked, bool join)

	CURLcode Curl_thrdq_create(struct curl_thrdq **ptqueue,
	const char *name,
-	uint32_t max_len,
	uint32_t min_threads,
	uint32_t max_threads,
	uint32_t idle_time_ms,
@@ -222,7 +220,6 @@ CURLcode Curl_thrdq_create(struct curl_thrdq **ptqueue,
	tqueue->fn_process = fn_process;
	tqueue->fn_event = fn_event;
	tqueue->fn_user_data = user_data;
-	tqueue->send_max_len = max_len;

	tqueue->name = curlx_strdup(name);
	if(!tqueue->name)
@@ -253,11 +250,18 @@ void Curl_thrdq_destroy(struct curl_thrdq *tqueue, bool join)
	thrdq_unlink(tqueue, TRUE, join);
	}

+	static uint32_t thrdq_get_signals(struct curl_thrdq *tqueue)
+	{
+	size_t qlen = Curl_llist_count(&tqueue->sendq);
+	return (qlen <= UINT32_MAX) ? (uint32_t)qlen : UINT32_MAX;
+	}
+
	CURLcode Curl_thrdq_send(struct curl_thrdq tqueue, void item,
	const char *description, timediff_t timeout_ms)
	{
-	CURLcode result = CURLE_AGAIN;
-	size_t signals = 0;
+	struct thrdq_item *qitem;
+	CURLcode result = CURLE_OK;
+	uint32_t signals = 0;

	Curl_mutex_acquire(&tqueue->lock);
	if(tqueue->aborted) {
@@ -270,19 +274,13 @@ CURLcode Curl_thrdq_send(struct curl_thrdq tqueue, void item,
	goto out;
	}

-	if(!tqueue->send_max_len \|\|
-	(Curl_llist_count(&tqueue->sendq) < tqueue->send_max_len)) {
-	struct thrdq_item *qitem = thrdq_item_create(tqueue, item, description,
-	timeout_ms);
-	if(!qitem) {
-	result = CURLE_OUT_OF_MEMORY;
-	goto out;
-	}
-	item = NULL;
-	Curl_llist_append(&tqueue->sendq, qitem, &qitem->node);
-	signals = Curl_llist_count(&tqueue->sendq);
-	result = CURLE_OK;
+	qitem = thrdq_item_create(tqueue, item, description, timeout_ms);
+	if(!qitem) {
+	result = CURLE_OUT_OF_MEMORY;
+	goto out;
	}
+	Curl_llist_append(&tqueue->sendq, qitem, &qitem->node);
+	signals = thrdq_get_signals(tqueue);

	out:
	Curl_mutex_release(&tqueue->lock);
@@ -310,7 +308,7 @@ CURLcode Curl_thrdq_recv(struct curl_thrdq tqueue, void *pitem)
	{
	CURLcode result = CURLE_AGAIN;
	struct Curl_llist_node *e;
-	size_t signals = 0;
+	uint32_t signals = 0;

	*pitem = NULL;
	Curl_mutex_acquire(&tqueue->lock);
@@ -329,7 +327,8 @@ CURLcode Curl_thrdq_recv(struct curl_thrdq tqueue, void *pitem)
	result = CURLE_OK;
	}
	else
-	signals = Curl_llist_count(&tqueue->sendq);
+	signals = thrdq_get_signals(tqueue);
+
	out:
	Curl_mutex_release(&tqueue->lock);
	/* Signal thread pool unlocked to avoid deadlocks. If items await
@@ -384,17 +383,15 @@ UNITTEST CURLcode thrdq_await_done(struct curl_thrdq *tqueue,
	#endif

	CURLcode Curl_thrdq_set_props(struct curl_thrdq *tqueue,
-	uint32_t max_len,
	uint32_t min_threads,
	uint32_t max_threads,
	uint32_t idle_time_ms)
	{
	CURLcode result;
-	size_t signals;
+	uint32_t signals;

	Curl_mutex_acquire(&tqueue->lock);
-	tqueue->send_max_len = max_len;
-	signals = Curl_llist_count(&tqueue->sendq);
+	signals = thrdq_get_signals(tqueue);
	Curl_mutex_release(&tqueue->lock);

	result = Curl_thrdpool_set_props(tqueue->tpool, min_threads,

lib/thrdqueue.h+0 -3

@@ -54,7 +54,6 @@ typedef void Curl_thrdq_item_free_cb(void *item);
	*/
	CURLcode Curl_thrdq_create(struct curl_thrdq **ptqueue,
	const char *name,
-	uint32_t max_len, /* 0 for unlimited */
	uint32_t min_threads,
	uint32_t max_threads,
	uint32_t idle_time_ms,
@@ -74,7 +73,6 @@ void Curl_thrdq_destroy(struct curl_thrdq *tqueue, bool join);
	* to "item" on success, e.g. the queue takes ownership.
	* `description` is an optional string describing the item for tracing
	* purposes. It needs to have the same lifetime as `item`.
-	* Returns CURLE_AGAIN when the queue has already been full.
	*
	* With`timeout_ms` != 0, items that get stuck that long in the send
	* queue are removed and added to the receive queue right away.
@@ -115,7 +113,6 @@ CURLcode Curl_thrdq_await_done(struct curl_thrdq *tqueue,
	uint32_t timeout_ms);

	CURLcode Curl_thrdq_set_props(struct curl_thrdq *tqueue,
-	uint32_t max_len, /* 0 for unlimited */
	uint32_t min_threads,
	uint32_t max_threads,
	uint32_t idle_time_ms);

tests/http/test_21_resolve.py+25 -0

@@ -26,6 +26,7 @@
	#
	import logging
	import os
+	import re
	from datetime import timedelta
	from typing import Generator

@@ -283,6 +284,30 @@ class TestResolve:
	r.check_stats(count=1, http_status=0, exitcode=6)
	assert r.duration < timedelta(seconds=20), f'{r}'

+	# dnsd with one answer for AAAA, delayed one for A
+	@pytest.mark.skipif(condition=not Env.curl_override_dns(), reason="no DNS override")
+	@pytest.mark.skipif(condition=not Env.curl_has_feature('IPv6'), reason="no IPv6")
+	@pytest.mark.skipif(condition=not Env.curl_resolv_threaded(), reason="no threaded resolver")
+	def test_21_16_dnsd_link_local(self, env: Env, httpd, dnsd):
+	dnsd.set_answers(addr_aaaa=['[fe80::1]'])
+	run_env = os.environ.copy()
+	run_env['CURL_DNS_SERVER'] = f'127.0.0.1:{dnsd.port}'
+	run_env['CURL_QUICK_EXIT'] = '1'
+	run_env['CURL_DEBUG'] = 'dns'
+	curl = CurlClient(env=env, run_env=run_env, force_resolv=False)
+	url = f'https://{env.authority_for(env.domain1, "http/1.1")}/data.json'
+	r = curl.http_download(urls=[url], with_stats=True, extra_args=[
+	'--connect-timeout', '1'
+	])
+	# should fail with CURLE_OPERATION_TIMEOUT or COULDNT_CONNECT
+	assert r.exit_code in (7, 28), f'{r.dump_logs()}'
+	af_unspec_resolves = [line for line in r.trace_lines if
+	re.match(r'.* \[DNS] re-queueing query .+ for AF_UNSPEC resolve', line)]
+	assert len(af_unspec_resolves) == 1, f'{r.dump_logs()}'
+	aaaa_resolves = [line for line in r.trace_lines if
+	re.match(r'.* \* IPv6: fe80::1', line)]
+	assert len(aaaa_resolves) == 1, f'{r.dump_logs()}'
+
	def _clean_files(self, files):
	for file in files:
	if os.path.exists(file):

tests/unit/unit3301.c+2 -2

@@ -88,7 +88,7 @@ static CURLcode test_unit3301(const char *arg)

	/* create and teardown queue */
	memset(&ctx, 0, sizeof(ctx));
-	result = Curl_thrdq_create(&tqueue, "unit3301-a", 0, 0, 2, 1,
+	result = Curl_thrdq_create(&tqueue, "unit3301-a", 0, 2, 1,
	unit3301_item_free, unit3301_process,
	unit3301_event, &ctx);
	fail_unless(!result, "queue-a create");
@@ -99,7 +99,7 @@ static CURLcode test_unit3301(const char *arg)
	/* create queue, have it process `count` items */
	count = 10;
	memset(&ctx, 0, sizeof(ctx));
-	result = Curl_thrdq_create(&tqueue, "unit3301-b", 0, 0, 2, 1,
+	result = Curl_thrdq_create(&tqueue, "unit3301-b", 0, 2, 1,
	unit3301_item_free, unit3301_process,
	unit3301_event, &ctx);
	fail_unless(!result, "queue-b create");

tests/unit/unit3306.c+1 -1

@@ -76,7 +76,7 @@ static CURLcode test_unit3306(const char *arg)
	fail_unless(getenv("CURL_DBG_THRDPOOL_FAIL_STARTS"),
	"CURL_DBG_THRDPOOL_FAIL_STARTS must be set for this test");

-	result = Curl_thrdq_create(&tqueue, "unit3306", 0, 0, 2, 1,
+	result = Curl_thrdq_create(&tqueue, "unit3306", 0, 2, 1,
	unit3306_item_free, unit3306_process,
	NULL, NULL);
	fail_unless(!result, "queue create");

Subject

Binary

Architecture & Compiler Flags

Build Log

Commit Diff GitHub →