romenskiy2012/curl
1
0
Fork 0
mirror of https://github.com/curl/curl.git synced 2026-08-25 17:53:43 +03:00
Code Issues Projects Releases Packages Wiki Activity

async-thrdd: use thread queue for resolving

Browse source 
Use a thread queue and pool for asnyc threaded DNS resolves.
Add pytest test_21_* for verification.

Add `CURLMOPT_RESOLVE_THREADS_MAX` to allow applications to
resize the thread pool used.

Add `CURLMOPT_QUICK_EXIT` to allow applications to skip thread
joins when cleaning up a multi handle. Multi handles in
`curl_easy_perform()` inherit this from `CURLOPT_QUICK_EXIT`.

Add several debug environment variables for testing.

Closes #20936
This commit is contained in:
Stefan Eissing 2026-03-24 12:50:53 +01:00 • committed by Daniel Stenberg
parent 507e7be573
commit 39036c9021
No known key found for this signature in database
GPG key ID: 5CC908FDB71E12C2
31 changed files with 998 additions and 614 deletions
Download patch file Download diff file Expand all files Collapse all files

940
lib/asyn-thrdd.c
View file

File diff suppressed because it is too large Load diff

53
lib/asyn.h
View file

@ -32,6 +32,7 @@
struct Curl_easy;
struct Curl_dns_entry;
struct Curl_resolv_async;
struct Curl_multi;
#ifdef CURLRES_ASYNCH
@ -169,39 +170,12 @@ CURLcode Curl_async_ares_set_dns_local_ip6(struct Curl_easy *data);
#endif /* USE_RESOLV_ARES */
#ifdef USE_RESOLV_THREADED
/* async resolving implementation using POSIX threads */
#include "curl_threads.h"
/* Context for threaded address resolver */
struct async_thrdd_addr_ctx {
curl_thread_t thread_hnd;
char *hostname; /* hostname to resolve, Curl_async.hostname
duplicate */
curl_mutex_t mutx;
#ifndef CURL_DISABLE_SOCKETPAIR
curl_socket_t sock_pair[2]; /* eventfd/pipes/socket pair */
#endif
struct Curl_addrinfo *res;
#ifdef HAVE_GETADDRINFO
struct addrinfo hints;
#endif
struct curltime start;
timediff_t interval_end;
unsigned int poll_interval;
int port;
int sock_error;
int ref_count;
BIT(thrd_done);
BIT(do_abort);
};
struct async_thrdd_item;
/* Context for threaded resolver */
struct async_thrdd_ctx {
/* `addr` is a pointer since this memory is shared with a started
* thread. Since threads cannot be killed, we use reference counting
* so that we can "release" our pointer to this memory while the
* thread is still running. */
struct async_thrdd_addr_ctx *addr;
struct async_thrdd_item *resolved;
#if defined(USE_HTTPSRR) && defined(USE_ARES)
struct {
ares_channel channel;
@ -210,6 +184,8 @@ struct async_thrdd_ctx {
BIT(done);
} rr;
#endif
BIT(queued);
BIT(done);
};
void Curl_async_thrdd_shutdown(struct Curl_easy *data,
@ -217,6 +193,18 @@ void Curl_async_thrdd_shutdown(struct Curl_easy *data,
void Curl_async_thrdd_destroy(struct Curl_easy *data,
struct Curl_resolv_async *async);
CURLcode Curl_async_thrdd_multi_init(struct Curl_multi *multi,
uint32_t min_threads,
uint32_t max_threads,
uint32_t idle_time_ms);
void Curl_async_thrdd_multi_destroy(struct Curl_multi *multi, bool join);
void Curl_async_thrdd_multi_process(struct Curl_multi *multi);
CURLcode Curl_async_thrdd_multi_set_props(struct Curl_multi *multi,
uint32_t min_threads,
uint32_t max_threads,
uint32_t idle_time_ms);
#endif /* USE_RESOLV_THREADED */
#ifndef CURL_DISABLE_DOH
@ -248,10 +236,15 @@ struct Curl_resolv_async {
#ifndef CURL_DISABLE_DOH
struct doh_probes *doh; /* DoH specific data for this request */
#endif
struct curltime start;
timediff_t interval_end;
timediff_t timeout_ms;
uint32_t poll_interval;
uint32_t id; /* unique id per easy handle of the resolve operation */
/* what is being resolved */
/* what is being resolved */
uint16_t port;
uint8_t ip_version;
uint8_t transport;
char hostname[1];
};

8
lib/cf-socket.c
View file

@ -530,7 +530,8 @@ CURLcode Curl_parse_interface(const char *input,
#ifndef CURL_DISABLE_BINDLOCAL
static CURLcode bindlocal(struct Curl_easy *data, struct connectdata *conn,
curl_socket_t sockfd, int af, unsigned int scope)
curl_socket_t sockfd, int af, unsigned int scope,
uint8_t transport)
{
struct Curl_sockaddr_storage sa;
struct sockaddr *sock = (struct sockaddr *)&sa; /* bind to this address */
@ -648,7 +649,7 @@ static CURLcode bindlocal(struct Curl_easy *data, struct connectdata *conn,
ip_version = CURL_IPRESOLVE_V6;
#endif
(void)Curl_resolv_blocking(data, host, 80, ip_version, &h);
(void)Curl_resolv_blocking(data, host, 80, ip_version, transport, &h);
if(h) {
int h_af = h->addr->ai_family;
/* convert the resolved address, sizeof myhost >= INET_ADDRSTRLEN */
@ -1143,7 +1144,8 @@ static CURLcode cf_socket_open(struct Curl_cfilter *cf,
#endif
) {
result = bindlocal(data, cf->conn, ctx->sock, ctx->addr.family,
Curl_ipv6_scope(&ctx->addr.curl_sa_addr));
Curl_ipv6_scope(&ctx->addr.curl_sa_addr),
ctx->transport);
if(result) {
if(result == CURLE_UNSUPPORTED_PROTOCOL) {
/* The address family is not supported on this interface.

12
lib/cfilters.c
View file

@ -700,6 +700,18 @@ unsigned char Curl_conn_get_transport(struct Curl_easy *data,
return Curl_conn_cf_get_transport(cf, data);
}
int Curl_socktype_for_transport(uint8_t transport)
{
switch(transport) {
case TRNSPRT_TCP:
return SOCK_STREAM;
case TRNSPRT_UNIX:
return SOCK_STREAM;
default: /* UDP and QUIC */
return SOCK_DGRAM;
}
}
const char *Curl_conn_get_alpn_negotiated(struct Curl_easy *data,
struct connectdata *conn)
{

2
lib/cfilters.h
View file

@ -345,6 +345,8 @@ bool Curl_conn_cf_needs_flush(struct Curl_cfilter *cf,
unsigned char Curl_conn_cf_get_transport(struct Curl_cfilter *cf,
struct Curl_easy *data);
int Curl_socktype_for_transport(uint8_t transport);
const char *Curl_conn_cf_get_alpn_negotiated(struct Curl_cfilter *cf,
struct Curl_easy *data);

3
lib/easy.c
View file

@ -779,8 +779,9 @@ static CURLcode easy_perform(struct Curl_easy *data, bool events)
if(multi->in_callback)
return CURLE_RECURSIVE_API_CALL;
/* Copy the MAXCONNECTS option to the multi handle */
/* Copy relevant easy options to the multi handle */
curl_multi_setopt(multi, CURLMOPT_MAXCONNECTS, (long)data->set.maxconnects);
curl_multi_setopt(multi, CURLMOPT_QUICK_EXIT, (long)data->set.quick_exit);
data->multi_easy = NULL; /* pretend it does not exist */
mresult = curl_multi_add_handle(multi, data);

6
lib/ftp.c
View file

@ -1063,6 +1063,7 @@ static CURLcode ftp_port_resolve_host(struct Curl_easy *data,
*resp = NULL;
result = Curl_resolv_blocking(data, host, 0, conn->ip_version,
Curl_conn_get_transport(data, conn),
dns_entryp);
if(result)
failf(data, "failed to resolve the address provided to PORT: %s", host);
@ -2163,6 +2164,7 @@ static CURLcode ftp_state_pasv_resp(struct Curl_easy *data,
(void)Curl_resolv_blocking(data, host_name, ipquad.remote_port,
is_ipv6 ? CURL_IPRESOLVE_V6 : CURL_IPRESOLVE_V4,
Curl_conn_get_transport(data, conn),
&dns);
/* we connect to the proxy's port */
connectport = (unsigned short)ipquad.remote_port;
@ -2187,7 +2189,9 @@ static CURLcode ftp_state_pasv_resp(struct Curl_easy *data,
goto error;
}
(void)Curl_resolv_blocking(data, newhost, newport, conn->ip_version, &dns);
(void)Curl_resolv_blocking(data, newhost, newport, conn->ip_version,
Curl_conn_get_transport(data, conn),
&dns);
connectport = newport; /* we connect to the remote port */
if(!dns) {

54
lib/hostip.c
View file

@ -373,7 +373,9 @@ static bool can_resolve_ip_version(struct Curl_easy *data, int ip_version)
static CURLcode hostip_async_new(struct Curl_easy *data,
const char *hostname,
uint16_t port,
uint8_t ip_version)
uint8_t ip_version,
uint8_t transport,
timediff_t timeout_ms)
{
struct Curl_resolv_async *async;
size_t hostlen = strlen(hostname);
@ -389,6 +391,9 @@ static CURLcode hostip_async_new(struct Curl_easy *data,
async->id = data->state.next_async_id++;
async->port = port;
async->ip_version = ip_version;
async->transport = transport;
async->start = *Curl_pgrs_now(data);
async->timeout_ms = timeout_ms;
if(hostlen)
memcpy(async->hostname, hostname, hostlen);
@ -401,6 +406,8 @@ static CURLcode hostip_resolv(struct Curl_easy *data,
const char *hostname,
uint16_t port,
uint8_t ip_version,
uint8_t transport,
timediff_t timeout_ms,
bool allowDOH,
struct Curl_dns_entry **entry)
{
@ -415,6 +422,8 @@ static CURLcode hostip_resolv(struct Curl_easy *data,
#ifdef USE_CURL_ASYNC
if(data->state.async)
Curl_async_destroy(data);
#else
(void)timeout_ms;
#endif
#ifndef CURL_DISABLE_DOH
@ -463,7 +472,8 @@ static CURLcode hostip_resolv(struct Curl_easy *data,
int st;
#ifdef CURLRES_ASYNCH
if(!data->state.async) {
result = hostip_async_new(data, hostname, port, ip_version);
result = hostip_async_new(data, hostname, port, ip_version,
transport, timeout_ms);
if(result)
goto error;
}
@ -501,7 +511,8 @@ static CURLcode hostip_resolv(struct Curl_easy *data,
#ifndef CURL_DISABLE_DOH
else if(!Curl_is_ipaddr(hostname) && allowDOH && data->set.doh) {
if(!data->state.async) {
result = hostip_async_new(data, hostname, port, ip_version);
result = hostip_async_new(data, hostname, port, ip_version,
transport, timeout_ms);
if(result)
goto error;
}
@ -518,7 +529,8 @@ static CURLcode hostip_resolv(struct Curl_easy *data,
#ifdef CURLRES_ASYNCH
if(!data->state.async) {
result = hostip_async_new(data, hostname, port, ip_version);
result = hostip_async_new(data, hostname, port, ip_version,
transport, timeout_ms);
if(result)
goto error;
}
@ -526,7 +538,7 @@ static CURLcode hostip_resolv(struct Curl_easy *data,
respwait = TRUE;
#else
respwait = FALSE; /* no async waiting here */
addr = Curl_sync_getaddrinfo(data, hostname, port, ip_version);
addr = Curl_sync_getaddrinfo(data, hostname, port, ip_version, transport);
if(addr)
result = CURLE_OK;
#endif
@ -577,13 +589,15 @@ CURLcode Curl_resolv_blocking(struct Curl_easy *data,
const char *hostname,
uint16_t port,
uint8_t ip_version,
uint8_t transport,
struct Curl_dns_entry **pdns)
{
CURLcode result;
DEBUGASSERT(hostname && *hostname);
*pdns = NULL;
/* We cannot do a blocking resolve using DoH currently */
result = hostip_resolv(data, hostname, port, ip_version, FALSE, pdns);
result = hostip_resolv(data, hostname, port, ip_version,
transport, 0, FALSE, pdns);
switch(result) {
case CURLE_OK:
DEBUGASSERT(*pdns);
@ -619,7 +633,8 @@ static CURLcode resolv_alarm_timeout(struct Curl_easy *data,
const char *hostname,
uint16_t port,
uint8_t ip_version,
timediff_t timeoutms,
uint8_t transport,
timediff_t timeout_ms,
struct Curl_dns_entry **entry)
{
#ifdef HAVE_SIGACTION
@ -636,14 +651,14 @@ static CURLcode resolv_alarm_timeout(struct Curl_easy *data,
CURLcode result;
DEBUGASSERT(hostname && *hostname);
DEBUGASSERT(timeoutms > 0);
DEBUGASSERT(timeout_ms > 0);
DEBUGASSERT(!data->set.no_signal);
#ifndef CURL_DISABLE_DOH
DEBUGASSERT(!data->set.doh);
#endif
*entry = NULL;
timeout = (timeoutms > LONG_MAX) ? LONG_MAX : (long)timeoutms;
timeout = (timeout_ms > LONG_MAX) ? LONG_MAX : (long)timeout_ms;
if(timeout < 1000) {
/* The alarm() function only provides integer second resolution, so if
we want to wait less than one second we must bail out already now. */
@ -696,7 +711,8 @@ static CURLcode resolv_alarm_timeout(struct Curl_easy *data,
/* Perform the actual name resolution. This might be interrupted by an
* alarm if it takes too long. */
result = hostip_resolv(data, hostname, port, ip_version, TRUE, entry);
result = hostip_resolv(data, hostname, port, ip_version, transport,
timeout_ms, TRUE, entry);
clean_up:
if(!prev_alarm)
@ -772,33 +788,35 @@ CURLcode Curl_resolv(struct Curl_easy *data,
const char *hostname,
uint16_t port,
uint8_t ip_version,
timediff_t timeoutms,
uint8_t transport,
timediff_t timeout_ms,
struct Curl_dns_entry **entry)
{
DEBUGASSERT(hostname && *hostname);
*entry = NULL;
if(timeoutms < 0)
if(timeout_ms < 0)
/* got an already expired timeout */
return CURLE_OPERATION_TIMEDOUT;
#ifdef USE_ALARM_TIMEOUT
if(timeoutms && data->set.no_signal) {
if(timeout_ms && data->set.no_signal) {
/* Cannot use ALARM when signals are disabled */
timeoutms = 0;
timeout_ms = 0;
}
if(timeoutms && !Curl_doh_wanted(data)) {
if(timeout_ms && !Curl_doh_wanted(data)) {
return resolv_alarm_timeout(data, hostname, port, ip_version,
timeoutms, entry);
transport, timeout_ms, entry);
}
#endif /* !USE_ALARM_TIMEOUT */
#ifndef CURLRES_ASYNCH
if(timeoutms)
if(timeout_ms)
infof(data, "timeout on name lookup is not supported");
#endif
return hostip_resolv(data, hostname, port, ip_version, TRUE, entry);
return hostip_resolv(data, hostname, port, ip_version, transport,
timeout_ms, TRUE, entry);
}

7
lib/hostip.h
View file

@ -91,13 +91,15 @@ CURLcode Curl_resolv(struct Curl_easy *data,
const char *hostname,
uint16_t port,
uint8_t ip_version,
timediff_t timeoutms,
uint8_t transport,
timediff_t timeout_ms,
struct Curl_dns_entry **pdns);
CURLcode Curl_resolv_blocking(struct Curl_easy *data,
const char *hostname,
uint16_t port,
uint8_t ip_version,
uint8_t transport,
struct Curl_dns_entry **pdns);
CURLcode Curl_resolv_timeout(struct Curl_easy *data,
@ -127,7 +129,8 @@ CURLcode Curl_resolver_error(struct Curl_easy *data, const char *detail);
struct Curl_addrinfo *Curl_sync_getaddrinfo(struct Curl_easy *data,
const char *hostname,
uint16_t port,
uint8_t ip_version);
uint8_t ip_version,
uint8_t transport);
#endif

4
lib/hostip4.c
View file

@ -70,11 +70,13 @@
struct Curl_addrinfo *Curl_sync_getaddrinfo(struct Curl_easy *data,
const char *hostname,
uint16_t port,
uint8_t ip_version)
uint8_t ip_version,
uint8_t transport)
{
struct Curl_addrinfo *ai = NULL;
(void)ip_version;
(void)transport;
ai = Curl_ipv4_resolve_r(hostname, port);
if(!ai)

6
lib/hostip6.c
View file

@ -65,7 +65,8 @@
struct Curl_addrinfo *Curl_sync_getaddrinfo(struct Curl_easy *data,
const char *hostname,
uint16_t port,
uint8_t ip_version)
uint8_t ip_version,
uint8_t transport)
{
struct addrinfo hints;
struct Curl_addrinfo *res;
@ -83,8 +84,7 @@ struct Curl_addrinfo *Curl_sync_getaddrinfo(struct Curl_easy *data,
memset(&hints, 0, sizeof(hints));
hints.ai_family = pf;
hints.ai_socktype =
(Curl_conn_get_transport(data, data->conn) == TRNSPRT_TCP) ?
hints.ai_socktype = (transport == TRNSPRT_TCP) ?
SOCK_STREAM : SOCK_DGRAM;
#ifndef USE_RESOLVE_ON_IPS

48
lib/multi.c
View file

@ -307,10 +307,24 @@ struct Curl_multi *Curl_multi_handle(uint32_t xfer_table_size,
goto error;
#endif
#ifdef USE_RESOLV_THREADED
if(xfer_table_size < CURL_XFER_TABLE_SIZE) { /* easy multi */
if(Curl_async_thrdd_multi_init(multi, 0, 2, 10))
goto error;
}
else { /* real multi handle */
if(Curl_async_thrdd_multi_init(multi, 0, 20, 2000))
goto error;
}
#endif
return multi;
error:
#ifdef USE_RESOLV_THREADED
Curl_async_thrdd_multi_destroy(multi, TRUE);
#endif
Curl_multi_ev_cleanup(multi);
Curl_hash_destroy(&multi->proto_hash);
Curl_dnscache_destroy(&multi->dnscache);
@ -2522,6 +2536,9 @@ static CURLMcode multi_runsingle(struct Curl_multi *multi,
Curl_uint32_bset_remove(&multi->dirty, data->mid);
if(data == multi->admin) {
#ifdef USE_RESOLV_THREADED
Curl_async_thrdd_multi_process(multi);
#endif
Curl_cshutdn_perform(&multi->cshutdn, multi->admin, sigpipe_ctx);
return CURLM_OK;
}
@ -2951,6 +2968,9 @@ CURLMcode curl_multi_cleanup(CURLM *m)
} while(Curl_uint32_tbl_next(&multi->xfers, mid, &mid, &entry));
}
#ifdef USE_RESOLV_THREADED
Curl_async_thrdd_multi_destroy(multi, !multi->quick_exit);
#endif
Curl_cpool_destroy(&multi->cpool);
Curl_cshutdn_destroy(&multi->cshutdn, multi->admin);
if(multi->admin) {
@ -3347,6 +3367,34 @@ CURLMcode curl_multi_setopt(CURLM *m, CURLMoption option, ...)
case CURLMOPT_NOTIFYDATA:
multi->ntfy.ntfy_cb_data = va_arg(param, void *);
break;
case CURLMOPT_RESOLVE_THREADS_MAX:
#ifdef USE_RESOLV_THREADED
uarg = va_arg(param, long);
if((uarg <= 0) || (uarg > UINT32_MAX))
mresult = CURLM_BAD_FUNCTION_ARGUMENT;
else {
CURLcode result = Curl_async_thrdd_multi_set_props(
multi, 0, (uint32_t)uarg, 2000);
switch(result) {
case CURLE_OK:
mresult = CURLM_OK;
break;
case CURLE_BAD_FUNCTION_ARGUMENT:
mresult = CURLM_BAD_FUNCTION_ARGUMENT;
break;
case CURLE_OUT_OF_MEMORY:
mresult = CURLM_OUT_OF_MEMORY;
break;
default:
mresult = CURLM_INTERNAL_ERROR;
break;
}
}
#endif
break;
case CURLMOPT_QUICK_EXIT:
multi->quick_exit = va_arg(param, long) ? 1 : 0;
break;
default:
mresult = CURLM_UNKNOWN_OPTION;
break;

4
lib/multihandle.h
View file

@ -109,6 +109,9 @@ struct Curl_multi {
struct Curl_dnscache dnscache; /* DNS cache */
struct Curl_ssl_scache *ssl_scache; /* TLS session pool */
#ifdef USE_RESOLV_THREADED
struct curl_thrdq *resolv_thrdq;
#endif
#ifdef USE_LIBPSL
/* PSL cache. */
@ -186,6 +189,7 @@ struct Curl_multi {
BIT(xfer_buf_borrowed); /* xfer_buf is currently being borrowed */
BIT(xfer_ulbuf_borrowed); /* xfer_ulbuf is currently being borrowed */
BIT(xfer_sockbuf_borrowed); /* xfer_sockbuf is currently being borrowed */
BIT(quick_exit); /* do not join threads on cleanup */
#ifdef DEBUGBUILD
BIT(warned); /* true after user warned of DEBUGBUILD */
#endif

8
lib/socks.c
View file

@ -324,7 +324,9 @@ static CURLproxycode socks4_resolving(struct socks_state *sx,
DEBUGASSERT(sx->hostname && *sx->hostname);
result = Curl_resolv(data, sx->hostname, sx->remote_port,
cf->conn->ip_version, 0, &dns);
cf->conn->ip_version,
Curl_conn_cf_get_transport(cf, data),
0, &dns);
if(result == CURLE_AGAIN) {
CURL_TRC_CF(data, cf, "SOCKS4 non-blocking resolve of %s", sx->hostname);
return CURLPX_OK;
@ -853,7 +855,9 @@ static CURLproxycode socks5_resolving(struct socks_state *sx,
DEBUGASSERT(sx->hostname && *sx->hostname);
result = Curl_resolv(data, sx->hostname, sx->remote_port,
cf->conn->ip_version, 0, &dns);
cf->conn->ip_version,
Curl_conn_cf_get_transport(cf, data),
0, &dns);
if(result == CURLE_AGAIN) {
CURL_TRC_CF(data, cf, "SOCKS5 non-blocking resolve of %s", sx->hostname);
return CURLPX_OK;

98
lib/thrdpool.c
View file

@ -121,7 +121,8 @@ static CURL_THREAD_RETURN_T CURL_STDCALL thrdslot_run(void *arg)
tpool->fn_return(item, tpool->aborted ? NULL : tpool->fn_user_data);
}
if(tpool->aborted)
if(tpool->aborted ||
(Curl_llist_count(&tpool->slots) > tpool->max_threads))
goto out;
tslot->idle = TRUE;
@ -235,6 +236,63 @@ static bool thrdpool_unlink(struct curl_thrdpool *tpool, bool locked)
return TRUE;
}
static CURLcode thrdpool_signal(struct curl_thrdpool *tpool,
uint32_t nthreads)
{
struct Curl_llist_node *e, *n;
CURLcode result = CURLE_OK;
DEBUGASSERT(!tpool->aborted);
thrdpool_join_zombies(tpool);
for(e = Curl_llist_head(&tpool->slots); e && nthreads; e = n) {
struct thrdslot *tslot = Curl_node_elem(e);
n = Curl_node_next(e);
if(tslot->idle) {
Curl_cond_signal(&tslot->await);
--nthreads;
}
else if(!tslot->starttime.tv_sec && !tslot->starttime.tv_usec) {
/* starting thread, queries for work soon. */
--nthreads;
}
}
while(nthreads && !result &&
Curl_llist_count(&tpool->slots) < tpool->max_threads) {
result = thrdslot_start(tpool);
if(result)
break;
--nthreads;
}
return result;
}
CURLcode Curl_thrdpool_set_props(struct curl_thrdpool *tpool,
uint32_t min_threads,
uint32_t max_threads,
uint32_t idle_time_ms)
{
CURLcode result = CURLE_OK;
size_t running;
if(!max_threads || (min_threads > max_threads))
return CURLE_BAD_FUNCTION_ARGUMENT;
Curl_mutex_acquire(&tpool->lock);
tpool->min_threads = min_threads;
tpool->max_threads = max_threads;
tpool->idle_time_ms = idle_time_ms;
running = Curl_llist_count(&tpool->slots);
if(tpool->min_threads > running) {
result = thrdpool_signal(tpool, tpool->min_threads - (uint32_t)running);
}
Curl_mutex_release(&tpool->lock);
return result;
}
CURLcode Curl_thrdpool_create(struct curl_thrdpool **ptpool,
const char *name,
uint32_t min_threads,
@ -257,9 +315,6 @@ CURLcode Curl_thrdpool_create(struct curl_thrdpool **ptpool,
Curl_cond_init(&tpool->await);
Curl_llist_init(&tpool->slots, NULL);
Curl_llist_init(&tpool->zombies, NULL);
tpool->min_threads = min_threads;
tpool->max_threads = max_threads;
tpool->idle_time_ms = idle_time_ms;
tpool->fn_take = fn_take;
tpool->fn_process = fn_process;
tpool->fn_return = fn_return;
@ -269,10 +324,8 @@ CURLcode Curl_thrdpool_create(struct curl_thrdpool **ptpool,
if(!tpool->name)
goto out;
if(tpool->min_threads)
result = Curl_thrdpool_signal(tpool, tpool->min_threads);
else
result = CURLE_OK;
result = Curl_thrdpool_set_props(tpool, min_threads, max_threads,
idle_time_ms);
out:
if(result && tpool) {
@ -316,35 +369,10 @@ void Curl_thrdpool_destroy(struct curl_thrdpool *tpool, bool join)
CURLcode Curl_thrdpool_signal(struct curl_thrdpool *tpool, uint32_t nthreads)
{
struct Curl_llist_node *e, *n;
CURLcode result = CURLE_OK;
CURLcode result;
Curl_mutex_acquire(&tpool->lock);
DEBUGASSERT(!tpool->aborted);
thrdpool_join_zombies(tpool);
for(e = Curl_llist_head(&tpool->slots); e && nthreads; e = n) {
struct thrdslot *tslot = Curl_node_elem(e);
n = Curl_node_next(e);
if(tslot->idle) {
Curl_cond_signal(&tslot->await);
--nthreads;
}
else if(!tslot->starttime.tv_sec && !tslot->starttime.tv_usec) {
/* starting thread, queries for work soon. */
--nthreads;
}
}
while(nthreads && !result &&
Curl_llist_count(&tpool->slots) < tpool->max_threads) {
result = thrdslot_start(tpool);
if(result)
break;
--nthreads;
}
result = thrdpool_signal(tpool, nthreads);
Curl_mutex_release(&tpool->lock);
return result;
}

6
lib/thrdpool.h
View file

@ -91,6 +91,12 @@ CURLcode Curl_thrdpool_signal(struct curl_thrdpool *tpool, uint32_t nthreads);
CURLcode Curl_thrdpool_await_idle(struct curl_thrdpool *tpool,
uint32_t timeout_ms);
/* Change the properties of a threadpool. */
CURLcode Curl_thrdpool_set_props(struct curl_thrdpool *tpool,
uint32_t min_threads,
uint32_t max_threads,
uint32_t idle_time_ms);
#ifdef CURLVERBOSE
void Curl_thrdpool_trace(struct curl_thrdpool *tpool,
struct Curl_easy *data,

31
lib/thrdqueue.c
View file

@ -282,16 +282,20 @@ CURLcode Curl_thrdq_send(struct curl_thrdq *tqueue, void *item,
result = CURLE_OUT_OF_MEMORY;
goto out;
}
item = NULL;
Curl_llist_append(&tqueue->sendq, qitem, &qitem->node);
result = CURLE_OK;
signals = Curl_llist_count(&tqueue->sendq);
result = CURLE_OK;
}
out:
Curl_mutex_release(&tqueue->lock);
/* Signal thread pool unlocked to avoid deadlocks */
/* Signal thread pool unlocked to avoid deadlocks. Since we added
* item to the queue already, it might have been taken for processing
* already. Any error in signalling the pool cannot be reported to
* the caller since it needs to give up ownership of item. */
if(!result && signals)
result = Curl_thrdpool_signal(tqueue->tpool, (uint32_t)signals);
(void)Curl_thrdpool_signal(tqueue->tpool, (uint32_t)signals);
return result;
}
@ -357,6 +361,27 @@ CURLcode Curl_thrdq_await_done(struct curl_thrdq *tqueue,
return Curl_thrdpool_await_idle(tqueue->tpool, timeout_ms);
}
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;
Curl_mutex_acquire(&tqueue->lock);
tqueue->send_max_len = max_len;
signals = Curl_llist_count(&tqueue->sendq);
Curl_mutex_release(&tqueue->lock);
result = Curl_thrdpool_set_props(tqueue->tpool, min_threads,
max_threads, idle_time_ms);
if(!result && signals)
result = Curl_thrdpool_signal(tqueue->tpool, (uint32_t)signals);
return result;
}
#ifdef CURLVERBOSE
void Curl_thrdq_trace(struct curl_thrdq *tqueue,
struct Curl_easy *data,

6
lib/thrdqueue.h
View file

@ -103,6 +103,12 @@ void Curl_thrdq_clear(struct curl_thrdq *tqueue,
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);
#ifdef CURLVERBOSE
void Curl_thrdq_trace(struct curl_thrdq *tqueue,
struct Curl_easy *data,

3
lib/url.c
View file

@ -3039,7 +3039,8 @@ static CURLcode resolve_server(struct Curl_easy *data,
}
result = Curl_resolv(data, ehost->name, eport,
conn->ip_version, timeout_ms, pdns);
conn->ip_version, conn->transport_wanted,
timeout_ms, pdns);
DEBUGASSERT(!result || !*pdns);
if(!result) { /* resolved right away, either sync or from dnscache */
DEBUGASSERT(*pdns);