thrdpool: retry failed thread starts while items wait

Verified in test 3306

Closes #22303

docs/libcurl/libcurl-env-dbg.md

@@ -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

@@ -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

@@ -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

@@ -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

@@ -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

@@ -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

@@ -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

@@ -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

@@ -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

@@ -0,0 +1,120 @@
+/***************************************************************************
+ *                                  _   _ ____  _
+ *  Project                     ___| | | |  _ \| |
+ *                             / __| | | | |_) | |
+ *                            | (__| |_| |  _ <| |___
+ *                             \___|\___/|_| \_\_____|
+ *
+ * Copyright (C) Daniel Stenberg, <daniel@haxx.se>, 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 */