From d6dd7c1ccb6cf304dcc9f9ad1279e208a8157bc5 Mon Sep 17 00:00:00 2001 From: Michael Vandeberg Date: Mon, 19 Oct 2026 11:00:58 -0600 Subject: [PATCH] fix: start only the missing threads after a failed pool start If a worker thread failed to start, the next post() started a full set of threads again, so the pool could run more threads than it was given. ensure_started() now starts only the missing ones. The test uses a new test/unit/fault harness, a cut-down copy of Corosio's, that replaces pthread_create to make one thread fail. It runs on Linux only. Closes #457 --- include/boost/capy/ex/thread_pool.hpp | 4 + src/ex/thread_pool.cpp | 4 +- test/unit/CMakeLists.txt | 5 + test/unit/Jamfile | 6 +- test/unit/fault/CMakeLists.txt | 38 +++++ test/unit/fault/Jamfile | 36 +++++ test/unit/fault/fault.hpp | 88 +++++++++++ test/unit/fault/fault_posix.cpp | 210 +++++++++++++++++++++++++ test/unit/fault/thread_pool_faults.cpp | 108 +++++++++++++ 9 files changed, 497 insertions(+), 2 deletions(-) create mode 100644 test/unit/fault/CMakeLists.txt create mode 100644 test/unit/fault/Jamfile create mode 100644 test/unit/fault/fault.hpp create mode 100644 test/unit/fault/fault_posix.cpp create mode 100644 test/unit/fault/thread_pool_faults.cpp diff --git a/include/boost/capy/ex/thread_pool.hpp b/include/boost/capy/ex/thread_pool.hpp index c9b5fc777..63506868e 100644 --- a/include/boost/capy/ex/thread_pool.hpp +++ b/include/boost/capy/ex/thread_pool.hpp @@ -266,6 +266,8 @@ class thread_pool::executor_type @throws std::system_error If this call posts and must start the pool's worker threads, and a thread cannot be started. `c` is not queued. + Threads that did start keep running, and a later call + starts only the missing ones. */ BOOST_CAPY_DECL std::coroutine_handle<> @@ -293,6 +295,8 @@ class thread_pool::executor_type @throws std::system_error If this call must start the pool's worker threads, and a thread cannot be started. `c` is not queued. + Threads that did start keep running, and a later call + starts only the missing ones. */ BOOST_CAPY_DECL void diff --git a/src/ex/thread_pool.cpp b/src/ex/thread_pool.cpp index 4528c7d19..05cd8826e 100644 --- a/src/ex/thread_pool.cpp +++ b/src/ex/thread_pool.cpp @@ -248,8 +248,10 @@ class thread_pool::impl std::lock_guard lock(mutex_); if(started_.load(std::memory_order_relaxed) || stop_) return; + // Start at the threads already running: if an earlier call + // threw partway, this one adds only the missing ones. threads_.reserve(num_threads_); - for(std::size_t i = 0; i < num_threads_; ++i) + for(std::size_t i = threads_.size(); i < num_threads_; ++i) threads_.emplace_back([this, i]{ run(i); }); started_.store(true, std::memory_order_release); } diff --git a/test/unit/CMakeLists.txt b/test/unit/CMakeLists.txt index 14774345b..8973d0638 100644 --- a/test/unit/CMakeLists.txt +++ b/test/unit/CMakeLists.txt @@ -8,6 +8,9 @@ # file(GLOB_RECURSE PFILES CONFIGURE_DEPENDS *.cpp *.hpp) + +# The fault harness shadows libc symbols; it links into its own target. +list(FILTER PFILES EXCLUDE REGEX "/fault/") list(APPEND PFILES CMakeLists.txt Jamfile) @@ -32,3 +35,5 @@ boost_capy_test_suite_discover_tests(boost_capy_tests) # Add the main test target to Boost's test suite add_dependencies(tests boost_capy_tests) + +add_subdirectory(fault) diff --git a/test/unit/Jamfile b/test/unit/Jamfile index 76418db8e..3581ed3fd 100644 --- a/test/unit/Jamfile +++ b/test/unit/Jamfile @@ -24,7 +24,9 @@ project darwin,norecover:static ; -for local f in [ glob-tree-ex . : *.cpp : file*.cpp ] +# `fault` excludes the whole directory: its shadows replace libc +# functions, so it builds its own executables (see fault/Jamfile). +for local f in [ glob-tree-ex . : *.cpp : file*.cpp fault ] { local parts = [ SPLIT_BY_CHARACTERS $(f:S=) : / ] ; local name = $(parts:J=_) ; @@ -57,3 +59,5 @@ for local f in [ glob-tree-ex . : file*.cpp ] off norecover:static ; } + +build-project fault ; diff --git a/test/unit/fault/CMakeLists.txt b/test/unit/fault/CMakeLists.txt new file mode 100644 index 000000000..59a59c0bd --- /dev/null +++ b/test/unit/fault/CMakeLists.txt @@ -0,0 +1,38 @@ +# +# Copyright (c) 2026 Michael Vandeberg +# +# Distributed under the Boost Software License, Version 1.0. (See accompanying +# file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt) +# +# Official repository: https://github.com/cppalliance/capy +# + +# The shadows replace libc functions for the whole program, so they +# link into their own executable. They are written against glibc; +# other platforms create threads differently and have no coverage here. +if(NOT CMAKE_SYSTEM_NAME STREQUAL "Linux") + return() +endif() + +file(GLOB FAULT_FILES CONFIGURE_DEPENDS *.cpp *.hpp) +list(APPEND FAULT_FILES CMakeLists.txt Jamfile) +source_group(TREE ${CMAKE_CURRENT_SOURCE_DIR} PREFIX "" FILES ${FAULT_FILES}) + +add_executable(boost_capy_fault_tests ${FAULT_FILES}) +target_link_libraries(boost_capy_fault_tests PRIVATE + Boost::capy_test_suite_main + Boost::capy + ${CMAKE_DL_LIBS}) + +# Interposition only works if the shadows land in the executable's +# dynamic symbol table. A superproject that builds everything with +# hidden visibility would strip them, and the readback then refuses +# to run rather than reporting faults that never fire. +set_target_properties(boost_capy_fault_tests PROPERTIES + C_VISIBILITY_PRESET default + CXX_VISIBILITY_PRESET default + VISIBILITY_INLINES_HIDDEN OFF + ENABLE_EXPORTS ON) + +boost_capy_test_suite_discover_tests(boost_capy_fault_tests) +add_dependencies(tests boost_capy_fault_tests) diff --git a/test/unit/fault/Jamfile b/test/unit/fault/Jamfile new file mode 100644 index 000000000..d458c7708 --- /dev/null +++ b/test/unit/fault/Jamfile @@ -0,0 +1,36 @@ +# +# Copyright (c) 2026 Michael Vandeberg +# +# Distributed under the Boost Software License, Version 1.0. (See accompanying +# file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt) +# +# Official repository: https://github.com/cppalliance/capy +# + +import ac ; +import os ; +import testing ; + +project + : requirements + $(c20-requires) + /boost/capy//boost_capy + ../../../extra/test_suite/test_main.cpp + ../../../extra/test_suite/test_suite.cpp + ../../../extra/test_suite + extra + on + linux:-ldl + # Interposition only works if the shadows land in the executable's + # dynamic symbol table; Boost builds tests with hidden visibility. + # so the flag stays on these targets instead of + # propagating into a private rebuild of capy. + global + ; + +# The shadows are written against glibc, so nothing is built anywhere +# else; the CMake side gates the same way. +if [ os.name ] = LINUX +{ + run thread_pool_faults.cpp fault_posix.cpp ; +} diff --git a/test/unit/fault/fault.hpp b/test/unit/fault/fault.hpp new file mode 100644 index 000000000..09696dcf2 --- /dev/null +++ b/test/unit/fault/fault.hpp @@ -0,0 +1,88 @@ +// +// Copyright (c) 2026 Michael Vandeberg +// +// Distributed under the Boost Software License, Version 1.0. (See accompanying +// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt) +// +// Official repository: https://github.com/cppalliance/capy +// + +#ifndef BOOST_CAPY_TEST_FAULT_HPP +#define BOOST_CAPY_TEST_FAULT_HPP + +/* A cut-down copy of Corosio's fault harness (test/unit/fault there): + the same fault_scope interface, with only the entry points Capy's + tests need and one arm per thread. +*/ + +namespace boost { +namespace capy { +namespace test { +namespace fault { + +/** OS entry points the harness can fail. + + Names equal the libc symbol. +*/ +enum class sys +{ + pthread_create +}; + +/** Fail one OS call on the current thread. + + While the scope is alive, the `nth` call to `which` made on this + thread returns its documented failure with error `err`. Calls + before and after it forward to the real function, and other + threads never observe the fault. An `nth` that is never reached + leaves an arm that only counts calls. + + @par Preconditions + No other `fault_scope` is alive on this thread; a second one + aborts. The scope is destroyed on the thread that created it. + + @par Example + @code + fault_scope f(sys::pthread_create, EAGAIN); + std::thread t([]{}); // throws std::system_error + @endcode +*/ +class fault_scope +{ +public: + /// Arm `which` to fail with `err` on its `nth` call. + fault_scope(sys which, int err, unsigned nth = 1); + + /// Disarm, even if the fault never fired. + ~fault_scope(); + + /// Return true once the armed call has been failed. + bool fired() const noexcept; + + /** Return how many calls to the armed symbol this arm has seen. + + @return Calls seen since the scope was constructed, including + the one that fired. + */ + unsigned count() const noexcept; + + fault_scope(fault_scope const&) = delete; + fault_scope& operator=(fault_scope const&) = delete; +}; + +/** Return true if a call to `which` reaches the harness. + + The harness checks at startup that the dynamic loader binds the + symbol to its replacement. A test asks here rather than asserting + on an arm that can never fire. + + @param which The entry point to ask about. +*/ +bool hook_is_live(sys which) noexcept; + +} // fault +} // test +} // capy +} // boost + +#endif diff --git a/test/unit/fault/fault_posix.cpp b/test/unit/fault/fault_posix.cpp new file mode 100644 index 000000000..5b9d68568 --- /dev/null +++ b/test/unit/fault/fault_posix.cpp @@ -0,0 +1,210 @@ +// +// Copyright (c) 2026 Michael Vandeberg +// +// Distributed under the Boost Software License, Version 1.0. (See accompanying +// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt) +// +// Official repository: https://github.com/cppalliance/capy +// + +#include "fault.hpp" + +#include +#include +#include +#include +#include +#include + +/* The replacement libc functions ("shadows") are defined in the test + executable. libstdc++ calls pthread_create through the dynamic loader, + which finds the executable's definition first. Each shadow asks + should_fail() and otherwise forwards to the real function, which it + finds with dlsym(RTLD_NEXT). + + A sanitizer runtime linked statically into the executable defines + its own pthread_create, and the shadow takes its place. Forwarding + to libc would then create threads the sanitizer never sees, so the + shadow forwards to the runtime's __interceptor_pthread_create, which + the runtime exports for this purpose. The reference is weak and is + null without a sanitizer. + + The readback at startup checks that the loader really binds + pthread_create to the shadow. Without it, a build that stripped the + shadow from the dynamic symbol table would leave every fault silently + dead. +*/ + +#if defined(__SANITIZE_ADDRESS__) || defined(__SANITIZE_THREAD__) +#define BOOST_CAPY_FAULT_SANITIZER 1 +#elif defined(__has_feature) +#if __has_feature(address_sanitizer) || __has_feature(thread_sanitizer) || \ + __has_feature(memory_sanitizer) +#define BOOST_CAPY_FAULT_SANITIZER 1 +#endif +#endif +#if !defined(BOOST_CAPY_FAULT_SANITIZER) +#define BOOST_CAPY_FAULT_SANITIZER 0 +#endif + +#if BOOST_CAPY_FAULT_SANITIZER +extern "C" void __interceptor_pthread_create() __attribute__((weak)); +#endif + +namespace boost { +namespace capy { +namespace test { +namespace fault { + +namespace { + +// The arm for this thread. Plain aggregate so the shadows touch +// nothing that could allocate or lock. +struct slot +{ + sys which = sys::pthread_create; + int err = 0; + unsigned nth = 1; + unsigned seen = 0; + bool armed = false; + bool fired = false; + // Tracks the owning fault_scope's lifetime, independent of `armed`: + // should_fail clears `armed` when the fault fires, but the scope + // keeps the arm until it is destroyed. + bool owned = false; +}; + +thread_local slot tls_arm; + +[[noreturn]] void +die(char const* msg) noexcept +{ + std::fputs(msg, stderr); + std::fputc('\n', stderr); + std::abort(); +} + +// Resolve the real entry point behind a shadow, or die naming it +// rather than crash inside the shadow. +void* +real_symbol(char const* name) noexcept +{ +#if BOOST_CAPY_FAULT_SANITIZER + void (*const interceptor)() = &__interceptor_pthread_create; + if(interceptor && std::strcmp(name, "pthread_create") == 0) + return reinterpret_cast(interceptor); +#endif + void* p = ::dlsym(RTLD_NEXT, name); + if(!p) + { + char msg[160]; + std::snprintf( + msg, sizeof(msg), + "fault harness: %s has no implementation behind the shadow", + name); + die(msg); + } + return p; +} + +// Return true when the current call must fail; the error is already +// in errno. +bool +should_fail(sys which) noexcept +{ + slot& s = tls_arm; + if(!s.armed || s.which != which) + return false; + if(++s.seen != s.nth) + return false; + s.armed = false; + s.fired = true; + errno = s.err; + return true; +} + +// Whether the loader binds pthread_create to the shadow. A null +// binding means there is no dynamic symbol to interpose, as in a +// static executable: the shadow is not live, and tests skip. Any +// other binding is a broken interposition and aborts. +bool const pthread_create_live = [] +{ + void* bound = ::dlsym(RTLD_DEFAULT, "pthread_create"); + void* hook = reinterpret_cast(&::pthread_create); + if(bound == hook) + return true; + if(!bound) + return false; + std::fprintf( + stderr, "fault harness: pthread_create is bound to %p, hook is %p\n", + bound, hook); + die("fault harness: the pthread_create shadow is not interposing"); +}(); + +} // namespace + +fault_scope::fault_scope(sys which, int err, unsigned nth) +{ + if(tls_arm.owned) + die("fault_scope: a fault is already armed on this thread"); + tls_arm = slot{}; + tls_arm.which = which; + tls_arm.err = err; + tls_arm.nth = nth; + tls_arm.armed = true; + tls_arm.owned = true; +} + +fault_scope::~fault_scope() +{ + tls_arm.armed = false; + tls_arm.owned = false; +} + +bool +fault_scope::fired() const noexcept +{ + return tls_arm.fired; +} + +unsigned +fault_scope::count() const noexcept +{ + return tls_arm.seen; +} + +bool +hook_is_live(sys which) noexcept +{ + switch(which) + { + case sys::pthread_create: + return pthread_create_live; + } + return false; +} + +} // fault +} // test +} // capy +} // boost + +// pthread_create reports through its return value, not errno. The +// armed error is returned directly, and std::thread's constructor +// turns it into a std::system_error. +extern "C" int +pthread_create( + pthread_t* t, + pthread_attr_t const* a, + void* (*fn)(void*), + void* arg) noexcept +{ + using namespace boost::capy::test::fault; + using real_t = int (*)( + pthread_t*, pthread_attr_t const*, void* (*)(void*), void*); + static auto const real = + reinterpret_cast(real_symbol("pthread_create")); + if(should_fail(sys::pthread_create)) + return errno ? errno : EAGAIN; + return real(t, a, fn, arg); +} diff --git a/test/unit/fault/thread_pool_faults.cpp b/test/unit/fault/thread_pool_faults.cpp new file mode 100644 index 000000000..d902e5fff --- /dev/null +++ b/test/unit/fault/thread_pool_faults.cpp @@ -0,0 +1,108 @@ +// +// Copyright (c) 2026 Michael Vandeberg +// +// Distributed under the Boost Software License, Version 1.0. (See accompanying +// file LICENSE_1_0.txt or copy at http://www.boost.org/LICENSE_1_0.txt) +// +// Official repository: https://github.com/cppalliance/capy +// + +#include + +#include + +#include "fault.hpp" +#include "test_suite.hpp" + +#include +#include +#include +#include + +namespace boost { +namespace capy { + +namespace { + +using test::fault::fault_scope; +using test::fault::hook_is_live; +using test::fault::sys; + +// An nth no test reaches: the arm only counts calls. +constexpr unsigned count_only = std::numeric_limits::max(); + +} // namespace + +// A worker thread that fails to start surfaces as a throw from post(), +// and the next post() starts only the workers that are missing. +struct thread_pool_faults_test +{ + // The second worker fails to start, so the retry starts one. + void + testRetryStartsOnlyMissingThreads() + { + // Declared first: join() can stop the pool with c2 still + // queued, and the destructor then abandons it. + continuation c1, c2; + c1.h = c2.h = std::noop_coroutine(); + thread_pool pool(2); + + { + fault_scope fault(sys::pthread_create, EAGAIN, 2); + BOOST_TEST_THROWS( + pool.get_executor().post(c1), std::system_error); + BOOST_TEST(fault.fired()); + } + { + fault_scope counter(sys::pthread_create, EAGAIN, count_only); + pool.get_executor().post(c2); + BOOST_TEST_EQ(counter.count(), 1u); + } + + pool.join(); + } + + // The first worker fails to start, so the retry starts both. + void + testRetryAfterFirstThreadFails() + { + // Declared first: join() can stop the pool with c2 still + // queued, and the destructor then abandons it. + continuation c1, c2; + c1.h = c2.h = std::noop_coroutine(); + thread_pool pool(2); + + { + fault_scope fault(sys::pthread_create, EAGAIN); + BOOST_TEST_THROWS( + pool.get_executor().post(c1), std::system_error); + BOOST_TEST(fault.fired()); + } + { + fault_scope counter(sys::pthread_create, EAGAIN, count_only); + pool.get_executor().post(c2); + BOOST_TEST_EQ(counter.count(), 2u); + } + + pool.join(); + } + + void + run() + { + if(!hook_is_live(sys::pthread_create)) + { + test_suite::log << "pthread_create hook not live; skipping\n"; + return; + } + testRetryStartsOnlyMissingThreads(); + testRetryAfterFirstThreadFails(); + } +}; + +TEST_SUITE( + thread_pool_faults_test, + "boost.capy.fault.thread_pool"); + +} // capy +} // boost