39161602895

Committed 18 Oct 2026 08:04PM UTC coverage: 86.03%. First build

Build # 39161602895

Build Type

Pull #460

github

Committed by

web-flow

Commit Message

Merge b7b7cf230 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%)

1946 of 2262 relevant lines covered (86.03%)

5408634.11 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,280,383 ✔
37	{
38	}	46,280,383 ✔
39
40	auto thread_pool::schedule_operation::await_suspend(std::coroutine_handle<> awaiting_coroutine) -> void	46,272,327 ✔
41	{
42	if (!m_thread_pool.enqueue(awaiting_coroutine, m_yield))	46,272,327 ✔
43	{
44	throw std::runtime_error("coro::thread_pool is shutting down, unable to schedule new tasks.");	1 ✔
45	}
46	}	46,300,088 ✔
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,280,992 ✔
73	{
74	if (!m_shutdown_requested.load(std::memory_order::acquire))	46,280,992 ✔
75	{
76	return schedule_operation{*this};	46,285,082 ✔
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,080,142 ✔
85	{
86	auto operation = schedule();	11,080,142 ✔
87	operation.m_yield = true;	11,087,169 ✔
88	return operation;	11,087,169 ✔
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,286,265 ✔
110	{
111	if (handle == nullptr \|\| handle.done())	26,286,265 ✔
112	{
113	return false;	×
114	}
115
116	return enqueue(handle, true);	26,282,293 ✔
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,716,261 ✔
152	if (++iterations == global_queue_interval)	84,716,261 ✔
153	{
154	iterations = 0;	1,390,534 ✔
155	handle = pop_global();	1,390,534 ✔
156	}
157	if (handle == nullptr)	84,716,282 ✔
158	{
159	handle = pop_local(idx);	83,752,582 ✔
160	}
161	if (handle == nullptr)	84,658,819 ✔
162	{
163	handle = pop_global();	66,135,939 ✔
164	}
165	if (handle == nullptr)	84,815,165 ✔
166	{
167	handle = steal(idx);	7,179,870 ✔
168	}
169
170	if (handle == nullptr)	84,772,312 ✔
171	{
172	std::unique_lock lk{m_wait_mutex};	7,150,495 ✔
173	m_sleeping.fetch_add(1, std::memory_order::relaxed);	7,159,985 ✔
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,159,985 ✔
179	handle = pop_local(idx);	7,159,982 ✔
180	if (handle == nullptr)	7,159,980 ✔
181	{
182	handle = pop_global();	7,159,985 ✔
183	}
184	if (handle == nullptr)	7,159,982 ✔
185	{
186	handle = steal(idx);	6,534,174 ✔
187	}
188	if (handle == nullptr)	7,159,984 ✔
189	{
190	if (stopping)	6,518,890 ✔
191	{
192	break;	201 ✔
193	}
194	m_wait_cv.wait(lk);	6,518,689 ✔
195	}
196	m_sleeping.fetch_sub(1, std::memory_order::relaxed);	7,159,783 ✔
197	}	7,159,984 ✔
198
199	if (handle != nullptr)	84,785,787 ✔
200	{
201	handle.resume();	78,256,920 ✔
202	}
203	}	84,716,060 ✔
204
205	t_current_executor = current_executor{};	77 ✔
206
207	if (m_opts.on_thread_stop_functor != nullptr)	77 ✔
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,551,996 ✔
214	{
215	if (!global && t_current_executor.m_thread_pool == this)	72,551,996 ✔
216	{
217	auto& local = m_worker_queues[t_current_executor.m_idx];	17,609,018 ✔
218	std::scoped_lock lk{local.m_mutex};	17,608,546 ✔
219	if (m_shutdown_requested.load(std::memory_order::acquire))	17,608,861 ✔
220	{
NEW 221	return false;	×
222	}
223	local.m_queue.emplace_back(handle);	17,608,564 ✔
224	}	35,217,118 ✔
225	else
226	{
227	std::scoped_lock lk{m_global_mutex};	54,942,978 ✔
228	if (m_shutdown_requested.load(std::memory_order::acquire))	55,098,697 ✔
229	{
230	return false;	3 ✔
231	}
232	m_global_queue.emplace_back(handle);	55,098,694 ✔
233	}	55,098,697 ✔
234
235	wake(1);	72,689,554 ✔
236	return true;	72,682,658 ✔
237	}
238
239	auto thread_pool::pop_local(std::size_t idx) noexcept -> std::coroutine_handle<>	90,856,822 ✔
240	{
241	auto& local = m_worker_queues[idx];	90,856,822 ✔
242	std::scoped_lock lk{local.m_mutex};	90,841,382 ✔
243	if (local.m_queue.empty())	90,844,280 ✔
244	{
245	return nullptr;	73,263,414 ✔
246	}
247
248	auto handle = local.m_queue.back();	17,570,592 ✔
249	local.m_queue.pop_back();	17,570,686 ✔
250	return handle;	17,570,330 ✔
251	}	90,797,093 ✔
252
253	auto thread_pool::pop_global() noexcept -> std::coroutine_handle<>	74,621,672 ✔
254	{
255	std::scoped_lock lk{m_global_mutex};	74,621,672 ✔
256	if (m_global_queue.empty())	74,806,900 ✔
257	{
258	return nullptr;	14,122,837 ✔
259	}
260
261	auto handle = m_global_queue.front();	60,696,391 ✔
262	m_global_queue.pop_front();	60,694,031 ✔
263	return handle;	60,694,211 ✔
264	}	74,817,043 ✔
265
266	auto thread_pool::steal(std::size_t idx) noexcept -> std::coroutine_handle<>	13,708,238 ✔
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,992,092 ✔
271	{
272	auto& victim = m_worker_queues[(idx + i) % m_opts.thread_count];	9,337,369 ✔
273	std::scoped_lock lk{victim.m_mutex};	9,309,413 ✔
274	if (!victim.m_queue.empty())	9,346,604 ✔
275	{
276	auto handle = victim.m_queue.front();	38,239 ✔
277	victim.m_queue.pop_front();	38,239 ✔
278	return handle;	38,239 ✔
279	}
280	}	9,327,661 ✔
281
282	return nullptr;	13,654,723 ✔
283	}
284
285	auto thread_pool::wake(std::size_t count) noexcept -> void	72,922,955 ✔
286	{
287	if (count == 0 \|\| m_sleeping.load(std::memory_order::relaxed) == 0)	145,844,883 ✔
288	{
289	return;	59,489,010 ✔
290	}
291
292	std::scoped_lock lk{m_wait_mutex};	13,432,109 ✔
293	if (count >= m_opts.thread_count)	13,431,439 ✔
294	{
295	m_wait_cv.notify_all();	6,069,013 ✔
296	}
297	else
298	{
299	for (std::size_t i = 0; i < count; ++i)	14,738,148 ✔
300	{
301	m_wait_cv.notify_one();	7,375,720 ✔
302	}
303	}
304	}	13,431,442 ✔
305
306	auto thread_pool::empty() const noexcept -> bool	800 ✔
307	{
308	std::scoped_lock lk{m_wait_mutex};	800 ✔
309	if (m_sleeping.load(std::memory_order::relaxed) != m_threads.size())	1,600 ✔
310	{
311	return false;	445 ✔
312	}
313	std::scoped_lock global_lk{m_global_mutex};	355 ✔
314	return m_global_queue.empty();	355 ✔
315	}	800 ✔
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 / 39161602895

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