diff --git a/lib/asyn-thrdd.c b/lib/asyn-thrdd.c index fa06550a58..0941c08f5e 100644 --- a/lib/asyn-thrdd.c +++ b/lib/asyn-thrdd.c @@ -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; + 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); + 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 + * + * 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); diff --git a/lib/asyn.h b/lib/asyn.h index 76996f888a..43e1dc6463 100644 --- a/lib/asyn.h +++ b/lib/asyn.h @@ -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; diff --git a/lib/cf-dns.c b/lib/cf-dns.c index 4152a00c17..e3693e3acc 100644 --- a/lib/cf-dns.c +++ b/lib/cf-dns.c @@ -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); } diff --git a/lib/thrdqueue.c b/lib/thrdqueue.c index f7da23f385..d1b5bcd646 100644 --- a/lib/thrdqueue.c +++ b/lib/thrdqueue.c @@ -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, diff --git a/lib/thrdqueue.h b/lib/thrdqueue.h index b6dd92a25d..0d84f6f6f1 100644 --- a/lib/thrdqueue.h +++ b/lib/thrdqueue.h @@ -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); diff --git a/tests/http/test_21_resolve.py b/tests/http/test_21_resolve.py index eaaf5c6006..940952400a 100644 --- a/tests/http/test_21_resolve.py +++ b/tests/http/test_21_resolve.py @@ -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): diff --git a/tests/unit/unit3301.c b/tests/unit/unit3301.c index 67b126200d..4ed9e81bd4 100644 --- a/tests/unit/unit3301.c +++ b/tests/unit/unit3301.c @@ -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"); diff --git a/tests/unit/unit3306.c b/tests/unit/unit3306.c index 5b26ba8244..85b587dda2 100644 --- a/tests/unit/unit3306.c +++ b/tests/unit/unit3306.c @@ -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");