← Commit list

4dc236a109

Graham Campbell
Graham Campbell
2026-07-21
SUCCESS Duration: 212s

Subject

thrdpool: retry failed thread starts while items wait

Binary

No binary (build failed)

Architecture & Compiler Flags

Architecture: x86_64-linux-gnu Compiler: gcc (Ubuntu 11.4.0-1ubuntu1~22.04.3) 11.4.0

—

Build Log

(log yok)

Commit Diff GitHub →

docs/libcurl/libcurl-env-dbg.md +5 -0lib/asyn-thrdd.c +9 -0lib/thrdpool.c +23 -1lib/thrdqueue.c +21 -0lib/thrdqueue.h +12 -1tests/data/Makefile.am +1 -1tests/data/test3306 +23 -0tests/http/test_21_resolve.py +18 -0tests/unit/Makefile.inc +2 -1tests/unit/unit3306.c +120 -0
docs/libcurl/libcurl-env-dbg.md+5 -0
@@ -197,6 +197,11 @@ eligible for negative DNS caching, instead of a transient failure.
Make libcurl fail a resolve for IPv6 only.
+## `CURL_DBG_THRDPOOL_FAIL_STARTS`
+
+Fail this many thread starts in a thread pool, as if the system
+refused to spawn more threads. Read when the pool is created.
+
## `CURL_QUICK_EXIT`
Make `curl` use the quick exit option, even when built in debug mode.
lib/asyn-thrdd.c+9 -0
@@ -708,6 +708,15 @@ CURLcode Curl_async_pollset(struct Curl_easy *data,
else
stutter_ms = 200;
timeout_ms = CURLMIN(stutter_ms, timeout_ms);
+#else
+ if(async->queries_ongoing &&
+ !Curl_thrdq_check_started(data->multi->resolv_thrdq)) {
+ /* The queue has items but starting a worker thread to process
+ them just failed again; expire soon to check once more,
+ instead of sleeping on the full resolve timeout. */
+ CURL_TRC_DNS(data, "resolver thread start failed again, retrying");
+ timeout_ms = CURLMIN(100, timeout_ms);
+ }
#endif
Curl_expire(data, timeout_ms, EXPIRE_ASYNC_NAME);
}
lib/thrdpool.c+23 -1
@@ -28,6 +28,7 @@
#include "llist.h"
#include "curl_threads.h"
#include "curlx/timeval.h"
+#include "curlx/strparse.h"
#include "thrdpool.h"
#ifdef CURLVERBOSE
#include "curl_trc.h"
@@ -64,6 +65,9 @@ struct curl_thrdpool {
uint32_t max_threads;
uint32_t idle_time_ms;
uint32_t next_id;
+#ifdef DEBUGBUILD
+ int dbg_fail_starts; /* fail this many thread starts */
+#endif
BIT(aborted);
BIT(detached);
};
@@ -166,7 +170,14 @@ static CURLcode thrdslot_start(struct curl_thrdpool *tpool)
tpool->refcount++;
tslot->running = TRUE;
- tslot->thread = Curl_thread_create(thrdslot_run, tslot);
+#ifdef DEBUGBUILD
+ if(tpool->dbg_fail_starts > 0) {
+ --tpool->dbg_fail_starts;
+ tslot->thread = curl_thread_t_null;
+ }
+ else
+#endif
+ tslot->thread = Curl_thread_create(thrdslot_run, tslot);
if(tslot->thread == curl_thread_t_null) { /* never started */
tslot->running = FALSE;
thrdpool_unlink(tpool, TRUE);
@@ -320,6 +331,17 @@ CURLcode Curl_thrdpool_create(struct curl_thrdpool **ptpool,
if(!tpool->name)
goto out;
+#ifdef DEBUGBUILD
+ {
+ const char *p = getenv("CURL_DBG_THRDPOOL_FAIL_STARTS");
+ if(p) {
+ curl_off_t l;
+ if(!curlx_str_number(&p, &l, INT_MAX))
+ tpool->dbg_fail_starts = (int)l;
+ }
+ }
+#endif
+
result = Curl_thrdpool_set_props(tpool, min_threads, max_threads,
idle_time_ms);
lib/thrdqueue.c+21 -0
@@ -295,10 +295,22 @@ out:
return result;
}
+bool Curl_thrdq_check_started(struct curl_thrdq *tqueue)
+{
+ size_t unprocessed;
+
+ Curl_mutex_acquire(&tqueue->lock);
+ unprocessed = tqueue->aborted ? 0 : Curl_llist_count(&tqueue->sendq);
+ Curl_mutex_release(&tqueue->lock);
+ return !unprocessed ||
+ !Curl_thrdpool_signal(tqueue->tpool, (uint32_t)unprocessed);
+}
+
CURLcode Curl_thrdq_recv(struct curl_thrdq *tqueue, void **pitem)
{
CURLcode result = CURLE_AGAIN;
struct Curl_llist_node *e;
+ size_t signals = 0;
*pitem = NULL;
Curl_mutex_acquire(&tqueue->lock);
@@ -316,8 +328,17 @@ CURLcode Curl_thrdq_recv(struct curl_thrdq *tqueue, void **pitem)
thrdq_item_destroy(qitem);
result = CURLE_OK;
}
+ else
+ signals = Curl_llist_count(&tqueue->sendq);
out:
Curl_mutex_release(&tqueue->lock);
+ /* Signal thread pool unlocked to avoid deadlocks. If items await
+ * processing while nothing was ready, make sure the pool has a
+ * thread to work on them. An earlier thread start may have failed,
+ * which `Curl_thrdq_send()` cannot report to its caller. Without
+ * this, such items would sit unprocessed until the next send. */
+ if(signals)
+ (void)Curl_thrdpool_signal(tqueue->tpool, (uint32_t)signals);
return result;
}
lib/thrdqueue.h+12 -1
@@ -86,10 +86,21 @@ CURLcode Curl_thrdq_send(struct curl_thrdq *tqueue, void *item,
* The caller takes ownership of the item received, e.g. the queue
* relinquishes all references to item.
* Returns CURLE_AGAIN when there is no processed item, setting `pitem`
- * to NULL.
+ * to NULL. When nothing has been processed while items await sending,
+ * the pool is signalled to make sure a worker thread exists: an
+ * earlier thread start may have failed.
*/
CURLcode Curl_thrdq_recv(struct curl_thrdq *tqueue, void **pitem);
+/* Check that items awaiting processing have a worker thread to run
+ * them, signalling the pool to start one when needed -- an earlier
+ * thread start may have failed. Returns TRUE when everything is ok:
+ * no items are waiting or the pool took the signal. FALSE when a
+ * thread start just failed again; callers may want to check again
+ * soon rather than wait indefinitely.
+ */
+bool Curl_thrdq_check_started(struct curl_thrdq *tqueue);
+
/* Return TRUE if the passed "item" matches. */
typedef bool Curl_thrdq_item_match_cb(void *item, void *match_data);
tests/data/Makefile.am+1 -1
@@ -286,7 +286,7 @@ test3208 test3209 test3210 test3211 test3212 test3213 test3214 test3215 \
test3216 test3217 test3218 test3219 test3220 test3221 test3222 \
test3223 test3224 test3225 test3226 test3227 test3228 test3229 \
\
-test3300 test3301 test3302 test3303 test3304 test3305 \
+test3300 test3301 test3302 test3303 test3304 test3305 test3306 \
\
test3400 test3401 \
\
tests/data/test3306+23 -0
@@ -0,0 +1,23 @@
+<?xml version="1.0" encoding="US-ASCII"?>
+<testcase>
+<info>
+<keywords>
+unittest
+threads
+</keywords>
+</info>
+
+# Client-side
+<client>
+<features>
+unittest
+Debug
+</features>
+<setenv>
+CURL_DBG_THRDPOOL_FAIL_STARTS=5
+</setenv>
+<name>
+thrdqueue retries failed thread starts on receive
+</name>
+</client>
+</testcase>
tests/http/test_21_resolve.py+18 -0
@@ -265,6 +265,24 @@ class TestResolve:
if env.curl_is_verbose():
assert not [t for t in r.trace_lines if 'Negative DNS entry' in t], f'{r}'
+ # a resolve gets processed even when the first resolver thread
+ # starts fail, e.g. when the system temporarily refuses to spawn
+ # threads. The transfer must fail on the (debug-forced) lookup
+ # failure well before the resolve timeout.
+ @pytest.mark.skipif(condition=not Env.curl_resolv_threaded(), reason="no threaded resolver")
+ def test_21_15_resolv_thread_start_fails(self, env: Env, httpd, nghttpx):
+ run_env = os.environ.copy()
+ run_env['CURL_DBG_THRDPOOL_FAIL_STARTS'] = '3'
+ run_env['CURL_DBG_RESOLV_FAIL_DELAY'] = '10'
+ curl = CurlClient(env=env, run_env=run_env, force_resolv=False)
+ url = 'https://test-1.http.curl.invalid/'
+ r = curl.http_download(urls=[url], with_stats=True, extra_args=[
+ '--connect-timeout', '20'
+ ])
+ r.check_exit_code(6)
+ r.check_stats(count=1, http_status=0, exitcode=6)
+ assert r.duration < timedelta(seconds=20), f'{r}'
+
def _clean_files(self, files):
for file in files:
if os.path.exists(file):
tests/unit/Makefile.inc+2 -1
@@ -50,4 +50,5 @@ TESTS_C = \
unit3200.c unit3205.c \
unit3211.c unit3212.c unit3213.c unit3214.c unit3216.c unit3219.c \
unit3227.c \
- unit3300.c unit3301.c unit3302.c unit3303.c unit3304.c unit3400.c
+ unit3300.c unit3301.c unit3302.c unit3303.c unit3304.c unit3306.c \
+ unit3400.c
tests/unit/unit3306.c+120 -0
@@ -0,0 +1,120 @@
+/***************************************************************************
+ * _ _ ____ _
+ * Project ___| | | | _ \| |
+ * / __| | | | |_) | |
+ * | (__| |_| | _ <| |___
+ * \___|\___/|_| \_\_____|
+ *
+ * Copyright (C) Daniel Stenberg, <[email protected]>, et al.
+ *
+ * This software is licensed as described in the file COPYING, which
+ * you should have received as part of this distribution. The terms
+ * are also available at https://curl.se/docs/copyright.html.
+ *
+ * You may opt to use, copy, modify, merge, publish, distribute and/or sell
+ * copies of the Software, and permit persons to whom the Software is
+ * furnished to do so, under the terms of the COPYING file.
+ *
+ * This software is distributed on an "AS IS" basis, WITHOUT WARRANTY OF ANY
+ * KIND, either express or implied.
+ *
+ * SPDX-License-Identifier: curl
+ *
+ ***************************************************************************/
+#include "unitcheck.h"
+
+#include "curlx/wait.h"
+#include "thrdqueue.h"
+#include "curl_threads.h"
+
+#if defined(USE_THREADS) && defined(DEBUGBUILD)
+
+struct unit3306_item {
+ int id;
+ BIT(processed);
+};
+
+static struct unit3306_item *unit3306_item_create(int id)
+{
+ struct unit3306_item *uitem;
+ uitem = curlx_calloc(1, sizeof(*uitem));
+ if(uitem) {
+ uitem->id = id;
+ curl_mfprintf(stderr, "created item %d\n", uitem->id);
+ }
+ return uitem;
+}
+
+static void unit3306_item_free(void *item)
+{
+ struct unit3306_item *uitem = item;
+ curl_mfprintf(stderr, "free item %d\n", uitem->id);
+ curlx_free(uitem);
+}
+
+static void unit3306_process(void *item)
+{
+ struct unit3306_item *uitem = item;
+ curlx_wait_ms(1);
+ uitem->processed = TRUE;
+}
+
+/* The test runs with CURL_DBG_THRDPOOL_FAIL_STARTS=5 set, making the
+ * pool's first 5 thread starts fail, as if the system temporarily ran
+ * against a thread limit. Each of the 3 sends attempts one thread
+ * start (3 failures) and no thread exists to process the queue. Only
+ * the receive side signalling the pool again - consuming the
+ * remaining 2 armed failures, one per empty receive, and then
+ * starting workers - lets the items get processed. */
+static CURLcode test_unit3306(const char *arg)
+{
+ UNITTEST_BEGIN_SIMPLE
+ struct curl_thrdq *tqueue;
+ int i, count = 3, nrecvd;
+ CURLcode result;
+
+ 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,
+ unit3306_item_free, unit3306_process,
+ NULL, NULL);
+ fail_unless(!result, "queue create");
+ for(i = 0; i < count; ++i) {
+ struct unit3306_item *uitem = unit3306_item_create(i);
+ fail_unless(uitem, "item create");
+ result = Curl_thrdq_send(tqueue, uitem, NULL, 0);
+ fail_unless(!result, "send");
+ }
+
+ /* no send was able to start a thread. Receive until all items come
+ back processed, which needs the recv side to signal the pool. */
+ nrecvd = 0;
+ for(i = 0; (nrecvd < count) && (i < 10000); ++i) {
+ void *item;
+ result = Curl_thrdq_recv(tqueue, &item);
+ fail_unless(!result || (result == CURLE_AGAIN), "recv");
+ if(item) {
+ struct unit3306_item *uitem = item;
+ curl_mfprintf(stderr, "received item %d\n", uitem->id);
+ ++nrecvd;
+ fail_unless(uitem->processed, "recv unprocessed item");
+ unit3306_item_free(item);
+ }
+ else
+ curlx_wait_ms(1);
+ }
+ Curl_thrdq_destroy(tqueue, TRUE);
+ tqueue = NULL;
+ fail_unless(nrecvd == count, "items not processed");
+
+ UNITTEST_END_SIMPLE
+}
+
+#else
+static CURLcode test_unit3306(const char *arg)
+{
+ UNITTEST_BEGIN_SIMPLE
+ UNITTEST_END_SIMPLE
+}
+#endif /* USE_THREADS && DEBUGBUILD */