39163047356

Committed 18 Oct 2026 08:22PM UTC coverage: 86.207%. First build

Build # 39163047356

Build Type

Pull #460

github

Committed by

web-flow

Commit Message

Merge 82d23321e into 277467ffd

Pull Request Pull Request #460: Issue 446/thread pool memory

Coverage Stats

131 of 144 new or added lines in 4 files covered. (90.97%)

1950 of 2262 relevant lines covered (86.21%)

5402203.58 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

89.94

/src/thread_pool.cpp

#include "coro/thread_pool.hpp"
#include "coro/detail/task_self_deleting.hpp"

#include <cstdint>

namespace coro
{

namespace detail
{
static auto
    make_spawned_joinable_wait_task(std::unique_ptr<coro::task_group<coro::thread_pool>> group_ptr) -> coro::task<void>
{
    co_await *group_ptr;
    co_return;
}

} // namespace detail

namespace
{
/// Identifies the executor thread running on this thread so `schedule()` can find its queue.
struct current_executor
{
    const thread_pool* m_thread_pool{nullptr};
    std::size_t        m_idx{0};
};

thread_local current_executor t_current_executor{};

/// Executors check the shared queue first once every this many iterations, so tasks resumed or scheduled from other
/// threads are not starved by long running fork-join work on the executor's own queue.
constexpr std::uint32_t global_queue_interval{61};
} // namespace

thread_pool::schedule_operation::schedule_operation(thread_pool& tp) noexcept : m_thread_pool(tp)
{
}

auto thread_pool::schedule_operation::await_suspend(std::coroutine_handle<> awaiting_coroutine) -> void
{
    if (!m_thread_pool.enqueue(awaiting_coroutine, m_yield))
    {
        throw std::runtime_error("coro::thread_pool is shutting down, unable to schedule new tasks.");
    }
}

thread_pool::thread_pool(options&& opts, private_constructor)
    : m_opts(opts),
      m_worker_queues(std::make_unique<worker_queue[]>(m_opts.thread_count))
{ m_threads.reserve(m_opts.thread_count); }

auto thread_pool::make_unique(options opts) -> std::unique_ptr<thread_pool>
{
    auto tp = std::make_unique<thread_pool>(std::move(opts), private_constructor{});

    // Initialize the background worker threads once the thread pool is fully constructed
    // so the workers have a full ready object to work with.
    for (uint32_t i = 0; i < tp->m_opts.thread_count; ++i)
    {
        tp->m_threads.emplace_back([tp = tp.get(), i]() { tp->executor(i); });
    }

    return tp;
}

thread_pool::~thread_pool()
{
    shutdown();
}

auto thread_pool::schedule() -> schedule_operation
{
    if (!m_shutdown_requested.load(std::memory_order::acquire))
    {
        return schedule_operation{*this};
    }
    else
    {
        throw std::runtime_error("coro::thread_pool is shutting down, unable to schedule new tasks.");
    }
}

auto thread_pool::yield() -> schedule_operation
{
    auto operation    = schedule();
    operation.m_yield = true;
    return operation;
}

auto thread_pool::spawn_detached(coro::task<void>&& task) noexcept -> bool
{
    auto wrapper_task = detail::make_task_self_deleting(std::move(task));
    auto handle       = wrapper_task.handle();
    if (!resume(handle))
    {
        handle.destroy();
        return false;
    }
    return true;
}

auto thread_pool::spawn_joinable(coro::task<void>&& task) noexcept -> coro::task<void>
{
    auto group_ptr = std::make_unique<coro::task_group<coro::thread_pool>>(this, std::move(task));
    return detail::make_spawned_joinable_wait_task(std::move(group_ptr));
}

auto thread_pool::resume(std::coroutine_handle<> handle) noexcept -> bool
{
    if (handle == nullptr || handle.done())
    {
        return false;
    }

    return enqueue(handle, true);
}

auto thread_pool::shutdown() noexcept -> void
{
    {
        std::scoped_lock lk{m_wait_mutex};
        if (m_shutdown_requested.exchange(true, std::memory_order::acq_rel))
        {
            return;
        }
        m_wait_cv.notify_all();
    }

    for (auto& thread : m_threads)
    {
        if (thread.joinable())
        {
            thread.join();
        }
    }
}

auto thread_pool::executor(std::size_t idx) -> void
{
    t_current_executor = current_executor{.m_thread_pool = this, .m_idx = idx};

    if (m_opts.on_thread_start_functor != nullptr)
    {
        m_opts.on_thread_start_functor(idx);
    }

    std::uint32_t iterations{0};
    while (true)
    {
        std::coroutine_handle<> handle{};
        if (++iterations == global_queue_interval)
        {
            iterations = 0;
            handle     = pop_global();
        }
        if (handle == nullptr)
        {
            handle = pop_local(idx);
        }
        if (handle == nullptr)
        {
            handle = pop_global();
        }
        if (handle == nullptr)
        {
            handle = steal(idx);
        }

        if (handle == nullptr)
        {
            std::unique_lock lk{m_wait_mutex};
            m_sleeping.fetch_add(1, std::memory_order::relaxed);

            // Register before rechecking every queue. A producer either publishes before its queue is checked,
            // or acquires that queue's mutex afterwards and observes this registration before taking m_wait_mutex.
            // Keep the registration on exit so empty() also accounts for stopped executors.
            const auto stopping = m_shutdown_requested.load(std::memory_order::acquire);
            handle = pop_local(idx);
            if (handle == nullptr)
            {
                handle = pop_global();
            }
            if (handle == nullptr)
            {
                handle = steal(idx);
            }
            if (handle == nullptr)
            {
                if (stopping)
                {
                    break;
                }
                m_wait_cv.wait(lk);
            }
            m_sleeping.fetch_sub(1, std::memory_order::relaxed);
        }

        if (handle != nullptr)
        {
            handle.resume();
        }
    }

    t_current_executor = current_executor{};

    if (m_opts.on_thread_stop_functor != nullptr)
    {
        m_opts.on_thread_stop_functor(idx);
    }
}

auto thread_pool::enqueue(std::coroutine_handle<> handle, bool global) noexcept -> bool
{
    if (!global && t_current_executor.m_thread_pool == this)
    {
        auto&            local = m_worker_queues[t_current_executor.m_idx];
        std::scoped_lock lk{local.m_mutex};
        if (m_shutdown_requested.load(std::memory_order::acquire))
        {
            return false;
        }
        local.m_queue.emplace_back(handle);
    }
    else
    {
        std::scoped_lock lk{m_global_mutex};
        if (m_shutdown_requested.load(std::memory_order::acquire))
        {
            return false;
        }
        m_global_queue.emplace_back(handle);
    }

    wake(1);
    return true;
}

auto thread_pool::pop_local(std::size_t idx) noexcept -> std::coroutine_handle<>
{
    auto&            local = m_worker_queues[idx];
    std::scoped_lock lk{local.m_mutex};
    if (local.m_queue.empty())
    {
        return nullptr;
    }

    auto handle = local.m_queue.back();
    local.m_queue.pop_back();
    return handle;
}

auto thread_pool::pop_global() noexcept -> std::coroutine_handle<>
{
    std::scoped_lock lk{m_global_mutex};
    if (m_global_queue.empty())
    {
        return nullptr;
    }

    auto handle = m_global_queue.front();
    m_global_queue.pop_front();
    return handle;
}

auto thread_pool::steal(std::size_t idx) noexcept -> std::coroutine_handle<>
{
    // Take a single task and go back to the own queue: draining a victim would expand its pending tasks breadth
    // first and keep all of their coroutine frames alive at once.
    for (std::size_t i = 1; i < m_opts.thread_count; ++i)
    {
        auto&            victim = m_worker_queues[(idx + i) % m_opts.thread_count];
        std::scoped_lock lk{victim.m_mutex};
        if (!victim.m_queue.empty())
        {
            auto handle = victim.m_queue.front();
            victim.m_queue.pop_front();
            return handle;
        }
    }

    return nullptr;
}

auto thread_pool::wake(std::size_t count) noexcept -> void
{
    if (count == 0 || m_sleeping.load(std::memory_order::relaxed) == 0)
    {
        return;
    }

    std::scoped_lock lk{m_wait_mutex};
    if (count >= m_opts.thread_count)
    {
        m_wait_cv.notify_all();
    }
    else
    {
        for (std::size_t i = 0; i < count; ++i)
        {
            m_wait_cv.notify_one();
        }
    }
}

auto thread_pool::empty() const noexcept -> bool
{
    std::scoped_lock lk{m_wait_mutex};
    if (m_sleeping.load(std::memory_order::relaxed) != m_threads.size())
    {
        return false;
    }
    std::scoped_lock global_lk{m_global_mutex};
    return m_global_queue.empty();
}

auto thread_pool::queue_empty() const noexcept -> bool
{
    std::scoped_lock global_lk{m_global_mutex};
    if (!m_global_queue.empty())
    {
        return false;
    }
    for (std::size_t idx = 0; idx < m_opts.thread_count; ++idx)
    {
        auto&            local = m_worker_queues[idx];
        std::scoped_lock lk{local.m_mutex};
        if (!local.m_queue.empty())
        {
            return false;
        }
    }
    return true;
}

} // namespace coro

1	#include "coro/thread_pool.hpp"
2	#include "coro/detail/task_self_deleting.hpp"
3
4	#include <cstdint>
5
6	namespace coro
7	{
8
9	namespace detail
10	{
11	static auto
12	make_spawned_joinable_wait_task(std::unique_ptr<coro::task_group<coro::thread_pool>> group_ptr) -> coro::task<void>	1 ✔
13	{
14	co_await *group_ptr;
15	co_return;
16	}	2 ✔
17
18	} // namespace detail
19
20	namespace
21	{
22	/// Identifies the executor thread running on this thread so `schedule()` can find its queue.
23	struct current_executor
24	{
25	const thread_pool* m_thread_pool{nullptr};
26	std::size_t m_idx{0};
27	};
28
29	thread_local current_executor t_current_executor{};
30
31	/// Executors check the shared queue first once every this many iterations, so tasks resumed or scheduled from other
32	/// threads are not starved by long running fork-join work on the executor's own queue.
33	constexpr std::uint32_t global_queue_interval{61};
34	} // namespace
35
36	thread_pool::schedule_operation::schedule_operation(thread_pool& tp) noexcept : m_thread_pool(tp)	46,276,527 ✔
37	{
38	}	46,276,527 ✔
39
40	auto thread_pool::schedule_operation::await_suspend(std::coroutine_handle<> awaiting_coroutine) -> void	46,269,039 ✔
41	{
42	if (!m_thread_pool.enqueue(awaiting_coroutine, m_yield))	46,269,039 ✔
43	{
44	throw std::runtime_error("coro::thread_pool is shutting down, unable to schedule new tasks.");	1 ✔
45	}
46	}	46,298,814 ✔
47
48	thread_pool::thread_pool(options&& opts, private_constructor)	115 ✔
49	: m_opts(opts),	115 ✔
50	m_worker_queues(std::make_unique<worker_queue[]>(m_opts.thread_count))	115 ✔
51	{ m_threads.reserve(m_opts.thread_count); }	115 ✔
52
53	auto thread_pool::make_unique(options opts) -> std::unique_ptr<thread_pool>	115 ✔
54	{
55	auto tp = std::make_unique<thread_pool>(std::move(opts), private_constructor{});	115 ✔
56
57	// Initialize the background worker threads once the thread pool is fully constructed
58	// so the workers have a full ready object to work with.
59	for (uint32_t i = 0; i < tp->m_opts.thread_count; ++i)	316 ✔
60	{
61	tp->m_threads.emplace_back([tp = tp.get(), i]() { tp->executor(i); });	402 ✔
62	}
63
64	return tp;	115 ✔
65	}	×
66
67	thread_pool::~thread_pool()	230 ✔
68	{
69	shutdown();	115 ✔
70	}	230 ✔
71
72	auto thread_pool::schedule() -> schedule_operation	46,277,658 ✔
73	{
74	if (!m_shutdown_requested.load(std::memory_order::acquire))	46,277,658 ✔
75	{
76	return schedule_operation{*this};	46,281,046 ✔
77	}
78	else
79	{
80	throw std::runtime_error("coro::thread_pool is shutting down, unable to schedule new tasks.");
81	}
82	}
83
84	auto thread_pool::yield() -> schedule_operation	11,078,032 ✔
85	{
86	auto operation = schedule();	11,078,032 ✔
87	operation.m_yield = true;	11,085,052 ✔
88	return operation;	11,085,052 ✔
89	}
90
91	auto thread_pool::spawn_detached(coro::task<void>&& task) noexcept -> bool	200,004 ✔
92	{
93	auto wrapper_task = detail::make_task_self_deleting(std::move(task));	200,004 ✔
94	auto handle = wrapper_task.handle();	200,004 ✔
95	if (!resume(handle))	200,004 ✔
96	{
97	handle.destroy();	1 ✔
98	return false;	1 ✔
99	}
100	return true;	200,003 ✔
101	}
102
103	auto thread_pool::spawn_joinable(coro::task<void>&& task) noexcept -> coro::task<void>	1 ✔
104	{
105	auto group_ptr = std::make_unique<coro::task_group<coro::thread_pool>>(this, std::move(task));	1 ✔
106	return detail::make_spawned_joinable_wait_task(std::move(group_ptr));	1 ✔
107	}	1 ✔
108
109	auto thread_pool::resume(std::coroutine_handle<> handle) noexcept -> bool	26,161,274 ✔
110	{
111	if (handle == nullptr \|\| handle.done())	26,161,274 ✔
112	{
113	return false;	×
114	}
115
116	return enqueue(handle, true);	26,178,366 ✔
117	}
118
119	auto thread_pool::shutdown() noexcept -> void	183 ✔
120	{
121	{
122	std::scoped_lock lk{m_wait_mutex};	183 ✔
123	if (m_shutdown_requested.exchange(true, std::memory_order::acq_rel))	183 ✔
124	{
125	return;	68 ✔
126	}
127	m_wait_cv.notify_all();	115 ✔
128	}	183 ✔
129
130	for (auto& thread : m_threads)	316 ✔
131	{
132	if (thread.joinable())	201 ✔
133	{
134	thread.join();	201 ✔
135	}
136	}
137	}
138
139	auto thread_pool::executor(std::size_t idx) -> void	201 ✔
140	{
141	t_current_executor = current_executor{.m_thread_pool = this, .m_idx = idx};	201 ✔
142
143	if (m_opts.on_thread_start_functor != nullptr)	201 ✔
144	{
145	m_opts.on_thread_start_functor(idx);	4 ✔
146	}
147
148	std::uint32_t iterations{0};	201 ✔
149	while (true)
150	{
151	std::coroutine_handle<> handle{};	84,625,469 ✔
152	if (++iterations == global_queue_interval)	84,625,469 ✔
153	{
154	iterations = 0;	1,389,217 ✔
155	handle = pop_global();	1,389,217 ✔
156	}
157	if (handle == nullptr)	84,625,491 ✔
158	{
159	handle = pop_local(idx);	83,670,492 ✔
160	}
161	if (handle == nullptr)	84,588,970 ✔
162	{
163	handle = pop_global();	66,061,868 ✔
164	}
165	if (handle == nullptr)	84,730,978 ✔
166	{
167	handle = steal(idx);	7,084,403 ✔
168	}
169
170	if (handle == nullptr)	84,691,444 ✔
171	{
172	std::unique_lock lk{m_wait_mutex};	7,057,769 ✔
173	m_sleeping.fetch_add(1, std::memory_order::relaxed);	7,065,166 ✔
174
175	// Register before rechecking every queue. A producer either publishes before its queue is checked,
176	// or acquires that queue's mutex afterwards and observes this registration before taking m_wait_mutex.
177	// Keep the registration on exit so empty() also accounts for stopped executors.
178	const auto stopping = m_shutdown_requested.load(std::memory_order::acquire);	7,065,166 ✔
179	handle = pop_local(idx);	7,065,165 ✔
180	if (handle == nullptr)	7,065,168 ✔
181	{
182	handle = pop_global();	7,065,171 ✔
183	}
184	if (handle == nullptr)	7,065,169 ✔
185	{
186	handle = steal(idx);	6,455,015 ✔
187	}
188	if (handle == nullptr)	7,065,162 ✔
189	{
190	if (stopping)	6,439,289 ✔
191	{
192	break;	201 ✔
193	}
194	m_wait_cv.wait(lk);	6,439,088 ✔
195	}
196	m_sleeping.fetch_sub(1, std::memory_order::relaxed);	7,064,958 ✔
197	}	7,065,159 ✔
198
199	if (handle != nullptr)	84,705,302 ✔
200	{
201	handle.resume();	78,245,537 ✔
202	}
203	}	84,625,268 ✔
204
205	t_current_executor = current_executor{};	33 ✔
206
207	if (m_opts.on_thread_stop_functor != nullptr)	33 ✔
208	{
NEW 209	m_opts.on_thread_stop_functor(idx);	×
210	}
211	}	201 ✔
212
213	auto thread_pool::enqueue(std::coroutine_handle<> handle, bool global) noexcept -> bool	72,421,515 ✔
214	{
215	if (!global && t_current_executor.m_thread_pool == this)	72,421,515 ✔
216	{
217	auto& local = m_worker_queues[t_current_executor.m_idx];	17,614,691 ✔
218	std::scoped_lock lk{local.m_mutex};	17,613,534 ✔
219	if (m_shutdown_requested.load(std::memory_order::acquire))	17,613,884 ✔
220	{
NEW 221	return false;	×
222	}
223	local.m_queue.emplace_back(handle);	17,613,483 ✔
224	}	35,226,833 ✔
225	else
226	{
227	std::scoped_lock lk{m_global_mutex};	54,806,824 ✔
228	if (m_shutdown_requested.load(std::memory_order::acquire))	55,093,323 ✔
229	{
230	return false;	3 ✔
231	}
232	m_global_queue.emplace_back(handle);	55,093,320 ✔
233	}	55,093,323 ✔
234
235	wake(1);	72,678,806 ✔
236	return true;	72,672,491 ✔
237	}
238
239	auto thread_pool::pop_local(std::size_t idx) noexcept -> std::coroutine_handle<>	90,689,698 ✔
240	{
241	auto& local = m_worker_queues[idx];	90,689,698 ✔
242	std::scoped_lock lk{local.m_mutex};	90,665,750 ✔
243	if (local.m_queue.empty())	90,677,829 ✔
244	{
245	return nullptr;	73,093,009 ✔
246	}
247
248	auto handle = local.m_queue.back();	17,575,872 ✔
249	local.m_queue.pop_back();	17,576,010 ✔
250	return handle;	17,575,414 ✔
251	}	90,622,595 ✔
252
253	auto thread_pool::pop_global() noexcept -> std::coroutine_handle<>	74,461,411 ✔
254	{
255	std::scoped_lock lk{m_global_mutex};	74,461,411 ✔
256	if (m_global_queue.empty())	74,623,985 ✔
257	{
258	return nullptr;	13,946,565 ✔
259	}
260
261	auto handle = m_global_queue.front();	60,690,151 ✔
262	m_global_queue.pop_front();	60,687,655 ✔
263	return handle;	60,687,800 ✔
264	}	74,634,353 ✔
265
266	auto thread_pool::steal(std::size_t idx) noexcept -> std::coroutine_handle<>	13,534,533 ✔
267	{
268	// Take a single task and go back to the own queue: draining a victim would expand its pending tasks breadth
269	// first and keep all of their coroutine frames alive at once.
270	for (std::size_t i = 1; i < m_opts.thread_count; ++i)	22,360,805 ✔
271	{
272	auto& victim = m_worker_queues[(idx + i) % m_opts.thread_count];	8,876,972 ✔
273	std::scoped_lock lk{victim.m_mutex};	8,840,400 ✔
274	if (!victim.m_queue.empty())	8,879,408 ✔
275	{
276	auto handle = victim.m_queue.front();	37,723 ✔
277	victim.m_queue.pop_front();	37,723 ✔
278	return handle;	37,723 ✔
279	}
280	}	8,866,645 ✔
281
282	return nullptr;	13,483,833 ✔
283	}
284
285	auto thread_pool::wake(std::size_t count) noexcept -> void	72,914,913 ✔
286	{
287	if (count == 0 \|\| m_sleeping.load(std::memory_order::relaxed) == 0)	145,830,281 ✔
288	{
289	return;	59,150,152 ✔
290	}
291
292	std::scoped_lock lk{m_wait_mutex};	13,763,488 ✔
293	if (count >= m_opts.thread_count)	13,764,803 ✔
294	{
295	m_wait_cv.notify_all();	6,540,680 ✔
296	}
297	else
298	{
299	for (std::size_t i = 0; i < count; ++i)	14,462,125 ✔
300	{
301	m_wait_cv.notify_one();	7,238,000 ✔
302	}
303	}
304	}	13,764,807 ✔
305
306	auto thread_pool::empty() const noexcept -> bool	818 ✔
307	{
308	std::scoped_lock lk{m_wait_mutex};	818 ✔
309	if (m_sleeping.load(std::memory_order::relaxed) != m_threads.size())	1,636 ✔
310	{
311	return false;	463 ✔
312	}
313	std::scoped_lock global_lk{m_global_mutex};	355 ✔
314	return m_global_queue.empty();	355 ✔
315	}	818 ✔
316
NEW 317	auto thread_pool::queue_empty() const noexcept -> bool	×
318	{
NEW 319	std::scoped_lock global_lk{m_global_mutex};	×
NEW 320	if (!m_global_queue.empty())	×
321	{
NEW 322	return false;	×
323	}
NEW 324	for (std::size_t idx = 0; idx < m_opts.thread_count; ++idx)	×
325	{
NEW 326	auto& local = m_worker_queues[idx];	×
NEW 327	std::scoped_lock lk{local.m_mutex};	×
NEW 328	if (!local.m_queue.empty())	×
329	{
NEW 330	return false;	×
331	}
NEW 332	}	×
NEW 333	return true;	×
334	}	×
335
336	} // namespace coro

jbaldwin / libcoro / 39163047356

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous