24791269108

Committed 22 Apr 2026 04:55PM UTC coverage: 87.019%. First build

Build # 24791269108

Build Type

Pull #450

github

Committed by

web-flow

Commit Message

Merge fe5abd28e into 277467ffd

Pull Request Pull Request #450: coro::thread_pool_ws (work stealing)

Coverage Stats

295 of 310 new or added lines in 17 files covered. (95.16%)

2105 of 2419 relevant lines covered (87.02%)

4460863.69 hits per line

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

92.57

/src/thread_pool_ws.cpp

#include "coro/thread_pool_ws.hpp"
#include "coro/task_group.hpp"

#include <iostream>
#include <mutex>

#ifdef LIBCORO_TSAN
#include <sanitizer/tsan_interface.h>
#endif

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

} // namespace detail

thread_local std::optional<uint32_t> thread_pool_ws::m_thread_pool_queue_idx{std::nullopt};

thread_pool_ws::worker_info::worker_info(thread_pool_ws& tp, uint32_t i) : m_thread_pool(tp), m_idx(i)
{
}

thread_pool_ws::thread_pool_ws(options opts, private_constructor) : m_opts(std::move(opts))
{
    m_workers.reserve(m_opts.thread_count);
}

auto thread_pool_ws::worker_info::start() -> void
{
    // Each worker's thread is only started after *all* worker_info structures are initialized
    // since they all reference each other's queues for stealing.
    m_thread = std::thread([this]() -> void { m_thread_pool.execute(m_idx); });
}

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

    for (uint32_t i = 0; i < tp->m_opts.thread_count; ++i)
    {
        tp->m_workers.emplace_back(std::make_unique<worker_info>(*tp, i));
    }

    for (auto& worker : tp->m_workers)
    {
        worker->start();
    }

    while (tp->m_workers_started.load(std::memory_order::acquire) != tp->m_opts.thread_count)
    {
        std::this_thread::yield();
    }

    return tp;
}

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

thread_pool_ws::schedule_operation::schedule_operation(thread_pool_ws& tp, bool allocated) noexcept
    : m_thread_pool(tp),
      m_allocated(allocated)
{
#ifdef LIBCORO_TSAN
    // TSAN thinks the non-atomic constructor is a data race with the loads of this atomic on other threads.
    __tsan_release(&m_allocated);
#endif
}

auto thread_pool_ws::schedule_operation::await_suspend(std::coroutine_handle<> awaiting_coroutine) noexcept -> void
{
    m_awaiting_coroutine.store(awaiting_coroutine, std::memory_order::release);
    auto& tp = m_thread_pool; // take a copy since allocated schedule_operations can be deleted before this function completes.

    // See if we are running on a thread pool worker to enqueue locally.
    auto& idx = thread_pool_ws::m_thread_pool_queue_idx;
    if (idx.has_value())
    {
        m_thread_pool.m_workers[idx.value()]->m_queue.emplace(this);
    }
    else
    {
        detail::awaiter_list_push(m_thread_pool.m_global_queue, this);
    }

    // At this point `this` is no longer safe to use since a worker thread could have completed this coroutine.
    tp.try_wake_worker();
}

auto thread_pool_ws::schedule() -> schedule_operation
{
    m_try_wake_workers_size.fetch_add(1, std::memory_order::release);
    m_queue_size.fetch_add(1, std::memory_order::release);
    m_size.fetch_add(1, std::memory_order::release);
    if (m_shutdown_requested.load(std::memory_order::acquire))
    {
        m_try_wake_workers_size.fetch_sub(1, std::memory_order::release);
        m_queue_size.fetch_sub(1, std::memory_order::release);
        m_size.fetch_sub(1, std::memory_order::release);
        throw std::runtime_error("coro::thread_pool_ws is shutting down, unable to schedule new tasks.");
    }

    return schedule_operation{*this};
}

auto thread_pool_ws::spawn_detached(coro::task<void> user_task) -> bool
{
    if (m_shutdown_requested.load(std::memory_order::acquire))
    {
        return false;
    }

    auto wrapper_task = detail::make_task_self_deleting(std::move(user_task));
    return resume(wrapper_task.handle());
}

auto thread_pool_ws::spawn_joinable(coro::task<void> user_task) -> coro::task<void>
{
    if (m_shutdown_requested.load(std::memory_order::acquire))
    {
        throw std::runtime_error("coro::thread_pool_ws is shutting down, unable to spawn new tasks.");
    }

    auto group_ptr = std::make_unique<coro::task_group<coro::thread_pool_ws>>(this, std::move(user_task));
    return detail::make_spawned_joinable_wait_task(std::move(group_ptr));
}

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

    m_try_wake_workers_size.fetch_add(1, std::memory_order::release);
    m_queue_size.fetch_add(1, std::memory_order::release);
    m_size.fetch_add(1, std::memory_order::release);
    if (m_shutdown_requested.load(std::memory_order::acquire))
    {
        m_try_wake_workers_size.fetch_sub(1, std::memory_order::release);
        m_queue_size.fetch_sub(1, std::memory_order::release);
        m_size.fetch_sub(1, std::memory_order::release);
        return false;
    }

    auto* op = new schedule_operation{*this, true};
    op->await_suspend(handle);
    return true;
}

auto thread_pool_ws::shutdown() noexcept -> void
{
    if (m_shutdown_requested.exchange(true, std::memory_order::acq_rel) == false)
    {
        {
            for (auto& info : m_workers)
            {
                std::unique_lock<std::mutex> lk{info->m_wait_mutex};
                info->m_sleeping.store(false, std::memory_order::release);
                info->m_wait_cv.notify_one();
            }
            // std::unique_lock<std::mutex> lk{m_wait_mutex};
            // m_wait_cv.notify_all();
        }

        while (m_workers_stopped.load(std::memory_order::acquire) < m_opts.thread_count)
        {
            std::this_thread::yield();
        }

        for (auto& worker : m_workers)
        {
            if (worker->m_thread.joinable())
            {
                worker->m_thread.join();
            }
        }

        // Make sure all try wake tasks are completed prior to destructing this thread pool.
        while (m_try_wake_workers_size.load(std::memory_order::acquire) > 0)
        {
            std::this_thread::yield();
        }
    }
}

auto thread_pool_ws::execute(uint32_t idx) -> void
{
    m_thread_pool_queue_idx = {idx};

    m_workers_started.fetch_add(1, std::memory_order::release);

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

    auto&  info         = m_workers[idx];
    auto&  thread_queue = info->m_queue;
    size_t spin_counter{0};
    while (!m_shutdown_requested.load(std::memory_order::acquire))
    {
        bool had_tasks = drain_thread_queue(thread_queue);
        had_tasks |= try_steal(idx);
        had_tasks |= drain_global_queue(thread_queue);

        if (had_tasks)
        {
            spin_counter = 0;
            continue;
        }

        if (++spin_counter <= 100)
        {
            continue;
        }

        // Notify to all scheduled tasks that this worker is going to sleep and upon
        // scheduling a task this worker will need to be woken up.
        {
            std::scoped_lock lk{m_wait_mutex, info->m_wait_mutex};
            info->m_sleeping.exchange(true);
            m_sleeping.fetch_add(1, std::memory_order::release);
        }

        // Go to sleep.
        {
            std::unique_lock lk{info->m_wait_mutex};
            info->m_wait_cv.wait(
                lk,
                [&]() {
                    return !info->m_sleeping.load(std::memory_order::acquire) ||
                           m_queue_size.load(std::memory_order::acquire) > 0 ||
                           m_shutdown_requested.load(std::memory_order::acquire);
                });
        }
    }

    // while true; do gdb -ex run -ex quit --args taskset -c 0-1 ./test/libcoro_test "[thread_pool_ws]"; done

    // Run until the queue is fully drained.
    while (m_queue_size.load(std::memory_order::acquire) > 0)
    {
        (void)drain_thread_queue(thread_queue);
        (void)try_steal(idx);
        (void)drain_global_queue(thread_queue);
    }

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

    m_workers_stopped.fetch_add(1, std::memory_order::release);
}

auto thread_pool_ws::drain_thread_queue(riften::Deque<schedule_operation*>& queue) -> bool
{
    bool had_task{false};
    while (!queue.empty())
    {
        had_task |= resume_task(queue.pop());
    }
    return had_task;
}

auto thread_pool_ws::try_steal(uint32_t my_idx) -> bool
{
    bool had_tasks{false};
    auto queue_size = m_workers.size();
    for (size_t i = 1; i < queue_size; ++i)
    {
        auto steal_idx = (my_idx + i) % queue_size;
        had_tasks |= drain_peer_queue(m_workers[steal_idx]->m_queue);
    }
    return had_tasks;
}

auto thread_pool_ws::drain_peer_queue(riften::Deque<schedule_operation*>& queue) -> bool
{
    bool had_task{false};
    while (!queue.empty())
    {
        had_task |= resume_task(queue.steal());
    }
    return had_task;
}

auto thread_pool_ws::drain_global_queue(riften::Deque<schedule_operation*>& queue) -> bool
{
    if (m_global_queue.load(std::memory_order::acquire) != nullptr)
    {
        auto* head_op = detail::awaiter_list_pop_all(m_global_queue);
        if (head_op != nullptr)
        {
            while (head_op != nullptr)
            {
                auto* next_op = head_op->m_next;
                queue.emplace(head_op);
                head_op = next_op;
            }
        }

        // 1. If we got the global list we have tasks.
        // 2. If we didn't get the global list we know we can possibly steal.
        // Either way there should be tasks to process.
        return true;
    }

    return false;
}

auto thread_pool_ws::resume_task(std::optional<schedule_operation*> op) -> bool
{
    if (op.has_value())
    {
        m_queue_size.fetch_sub(1, std::memory_order::release);
        auto v = op.value();
        auto allocated = v->m_allocated.load(std::memory_order::acquire);
        v->m_awaiting_coroutine.load(std::memory_order::acquire).resume();
        m_size.fetch_sub(1, std::memory_order::release);
        if (allocated)
        {
            delete v;
        }
        return true;
    }
    return false;
}

auto thread_pool_ws::try_wake_worker() noexcept -> void
{
    // We're shutting down, no need to even attempt to wake any workers.
    if (m_shutdown_requested.load(std::memory_order::acquire))
    {
        m_try_wake_workers_size.fetch_sub(1, std::memory_order::release);
        return;
    }

    auto asleep = m_sleeping.load(std::memory_order::acquire);
    while (asleep > 0 && !m_shutdown_requested.load(std::memory_order::acquire))
    {
        if (m_sleeping.compare_exchange_weak(
                asleep, asleep - 1, std::memory_order::acq_rel, std::memory_order::acquire))
        {
            for (auto & info : m_workers)
            {
                std::unique_lock lk {info->m_wait_mutex};
                if (info->m_sleeping.load(std::memory_order::acquire))
                {
                    info->m_sleeping.store(false, std::memory_order::release);
                    info->m_wait_cv.notify_one();
                    break; // for
                }
            }
            break; // while
        }
    }

    m_try_wake_workers_size.fetch_sub(1, std::memory_order::release);

    // // Attempt to wake a sleeper if there are any.
    // if (m_sleeping.load(std::memory_order::acquire) > 0)
    // {
    //     std::unique_lock<std::mutex> lk{m_wait_mutex};
    //     // Check again after acquiring the lock to see if there are any sleeping workers.
    //     if (m_sleeping.load(std::memory_order::acquire) == 0)
    //     {
    //         return;
    //     }

    //     m_sleeping.fetch_sub(1, std::memory_order::release);
    //     m_wait_cv.notify_one();
    // }
}

} // namespace coro

1	#include "coro/thread_pool_ws.hpp"
2	#include "coro/task_group.hpp"
3
4	#include <iostream>
5	#include <mutex>
6
7	#ifdef LIBCORO_TSAN
8	#include <sanitizer/tsan_interface.h>
9	#endif
10
11	namespace coro
12	{
13	namespace detail
14	{
15	static auto make_spawned_joinable_wait_task(std::unique_ptr<coro::task_group<coro::thread_pool_ws>> group_ptr)	8✔
16	-> coro::task<void>
17	{
18	co_await *group_ptr;
19	co_return;
20	}	16✔
21
22	} // namespace detail
23
24	thread_local std::optional<uint32_t> thread_pool_ws::m_thread_pool_queue_idx{std::nullopt};
25
26	thread_pool_ws::worker_info::worker_info(thread_pool_ws& tp, uint32_t i) : m_thread_pool(tp), m_idx(i)	28✔
27	{
28	}	28✔
29
30	thread_pool_ws::thread_pool_ws(options opts, private_constructor) : m_opts(std::move(opts))	15✔
31	{
32	m_workers.reserve(m_opts.thread_count);	15✔
33	}	15✔
34
35	auto thread_pool_ws::worker_info::start() -> void	28✔
36	{
37	// Each worker's thread is only started after all worker_info structures are initialized
38	// since they all reference each other's queues for stealing.
39	m_thread = std::thread([this]() -> void { m_thread_pool.execute(m_idx); });	55✔
40	}	28✔
41
42	auto thread_pool_ws::make_unique(options opts) -> std::unique_ptr<thread_pool_ws>	15✔
43	{
44	auto tp = std::make_unique<thread_pool_ws>(std::move(opts), private_constructor{});	15✔
45
46	for (uint32_t i = 0; i < tp->m_opts.thread_count; ++i)	43✔
47	{
48	tp->m_workers.emplace_back(std::make_unique<worker_info>(*tp, i));	28✔
49	}
50
51	for (auto& worker : tp->m_workers)	43✔
52	{
53	worker->start();	28✔
54	}
55
56	while (tp->m_workers_started.load(std::memory_order::acquire) != tp->m_opts.thread_count)	242✔
57	{
58	std::this_thread::yield();	106✔
59	}
60
61	return tp;	15✔
NEW 62	}	×
63
64	thread_pool_ws::~thread_pool_ws()	15✔
65	{
66	shutdown();	15✔
67	}	15✔
68
69	thread_pool_ws::schedule_operation::schedule_operation(thread_pool_ws& tp, bool allocated) noexcept	4,488,356✔
70	: m_thread_pool(tp),	4,524,947✔
71	m_allocated(allocated)	4,488,356✔
72	{
73	#ifdef LIBCORO_TSAN
74	// TSAN thinks the non-atomic constructor is a data race with the loads of this atomic on other threads.
75	__tsan_release(&m_allocated);
76	#endif
77	}	4,562,942✔
78
79	auto thread_pool_ws::schedule_operation::await_suspend(std::coroutine_handle<> awaiting_coroutine) noexcept -> void	4,379,429✔
80	{
81	m_awaiting_coroutine.store(awaiting_coroutine, std::memory_order::release);	4,379,429✔
82	auto& tp = m_thread_pool; // take a copy since allocated schedule_operations can be deleted before this function completes.	4,918,362✔
83
84	// See if we are running on a thread pool worker to enqueue locally.
85	auto& idx = thread_pool_ws::m_thread_pool_queue_idx;	4,918,362✔
86	if (idx.has_value())	4,918,362✔
87	{
88	m_thread_pool.m_workers[idx.value()]->m_queue.emplace(this);	4,464,711✔
89	}
90	else
91	{
92	detail::awaiter_list_push(m_thread_pool.m_global_queue, this);	301,044✔
93	}
94
95	// At this point `this` is no longer safe to use since a worker thread could have completed this coroutine.
96	tp.try_wake_worker();	4,926,191✔
97	}	4,867,943✔
98
99	auto thread_pool_ws::schedule() -> schedule_operation	4,033,217✔
100	{
101	m_try_wake_workers_size.fetch_add(1, std::memory_order::release);	4,033,217✔
102	m_queue_size.fetch_add(1, std::memory_order::release);	4,033,217✔
103	m_size.fetch_add(1, std::memory_order::release);	4,033,217✔
104	if (m_shutdown_requested.load(std::memory_order::acquire))	4,033,217✔
105	{
106	m_try_wake_workers_size.fetch_sub(1, std::memory_order::release);	1✔
107	m_queue_size.fetch_sub(1, std::memory_order::release);	1✔
108	m_size.fetch_sub(1, std::memory_order::release);	1✔
109	throw std::runtime_error("coro::thread_pool_ws is shutting down, unable to schedule new tasks.");	1✔
110	}
111
112	return schedule_operation{*this};	4,389,967✔
113	}
114
115	auto thread_pool_ws::spawn_detached(coro::task<void> user_task) -> bool	200,003✔
116	{
117	if (m_shutdown_requested.load(std::memory_order::acquire))	200,003✔
118	{
NEW 119	return false;	×
120	}
121
122	auto wrapper_task = detail::make_task_self_deleting(std::move(user_task));	200,003✔
123	return resume(wrapper_task.handle());	200,003✔
124	}
125
126	auto thread_pool_ws::spawn_joinable(coro::task<void> user_task) -> coro::task<void>	8✔
127	{
128	if (m_shutdown_requested.load(std::memory_order::acquire))	8✔
129	{
NEW 130	throw std::runtime_error("coro::thread_pool_ws is shutting down, unable to spawn new tasks.");	×
131	}
132
133	auto group_ptr = std::make_unique<coro::task_group<coro::thread_pool_ws>>(this, std::move(user_task));	8✔
134	return detail::make_spawned_joinable_wait_task(std::move(group_ptr));	16✔
135	}	8✔
136
137	auto thread_pool_ws::resume(std::coroutine_handle<> handle) noexcept -> bool	200,012✔
138	{
139	if (handle == nullptr \|\| handle.done())	200,012✔
140	{
NEW 141	return false;	×
142	}
143
144	m_try_wake_workers_size.fetch_add(1, std::memory_order::release);	200,012✔
145	m_queue_size.fetch_add(1, std::memory_order::release);	200,012✔
146	m_size.fetch_add(1, std::memory_order::release);	200,012✔
147	if (m_shutdown_requested.load(std::memory_order::acquire))	200,012✔
148	{
NEW 149	m_try_wake_workers_size.fetch_sub(1, std::memory_order::release);	×
NEW 150	m_queue_size.fetch_sub(1, std::memory_order::release);	×
NEW 151	m_size.fetch_sub(1, std::memory_order::release);	×
NEW 152	return false;	×
153	}
154
155	auto* op = new schedule_operation{*this, true};	200,012✔
156	op->await_suspend(handle);	200,012✔
157	return true;	200,012✔
158	}
159
160	auto thread_pool_ws::shutdown() noexcept -> void	18✔
161	{
162	if (m_shutdown_requested.exchange(true, std::memory_order::acq_rel) == false)	18✔
163	{
164	{
165	for (auto& info : m_workers)	43✔
166	{
167	std::unique_lock<std::mutex> lk{info->m_wait_mutex};	28✔
168	info->m_sleeping.store(false, std::memory_order::release);	28✔
169	info->m_wait_cv.notify_one();	28✔
170	}	28✔
171	// std::unique_lock<std::mutex> lk{m_wait_mutex};
172	// m_wait_cv.notify_all();
173	}
174
175	while (m_workers_stopped.load(std::memory_order::acquire) < m_opts.thread_count)	234✔
176	{
177	std::this_thread::yield();	102✔
178	}
179
180	for (auto& worker : m_workers)	43✔
181	{
182	if (worker->m_thread.joinable())	28✔
183	{
184	worker->m_thread.join();	28✔
185	}
186	}
187
188	// Make sure all try wake tasks are completed prior to destructing this thread pool.
189	while (m_try_wake_workers_size.load(std::memory_order::acquire) > 0)	30✔
190	{
NEW 191	std::this_thread::yield();	×
192	}
193	}
194	}	18✔
195
196	auto thread_pool_ws::execute(uint32_t idx) -> void	27✔
197	{
198	m_thread_pool_queue_idx = {idx};	27✔
199
200	m_workers_started.fetch_add(1, std::memory_order::release);	28✔
201
202	if (m_opts.on_thread_start_functor != nullptr)	28✔
203	{
NEW 204	m_opts.on_thread_start_functor(idx);	×
205	}
206
207	auto& info = m_workers[idx];	28✔
208	auto& thread_queue = info->m_queue;	28✔
209	size_t spin_counter{0};	28✔
210	while (!m_shutdown_requested.load(std::memory_order::acquire))	870,279✔
211	{
212	bool had_tasks = drain_thread_queue(thread_queue);	870,183✔
213	had_tasks \|= try_steal(idx);	870,181✔
214	had_tasks \|= drain_global_queue(thread_queue);	870,180✔
215
216	if (had_tasks)	870,252✔
217	{
218	spin_counter = 0;	349,577✔
219	continue;	349,577✔
220	}
221
222	if (++spin_counter <= 100)	520,675✔
223	{
224	continue;	520,596✔
225	}
226
227	// Notify to all scheduled tasks that this worker is going to sleep and upon
228	// scheduling a task this worker will need to be woken up.
229	{
230	std::scoped_lock lk{m_wait_mutex, info->m_wait_mutex};	79✔
231	info->m_sleeping.exchange(true);	79✔
232	m_sleeping.fetch_add(1, std::memory_order::release);	79✔
233	}	79✔
234
235	// Go to sleep.
236	{
237	std::unique_lock lk{info->m_wait_mutex};	79✔
238	info->m_wait_cv.wait(	79✔
239	lk,
240	[&]() {	145✔
241	return !info->m_sleeping.load(std::memory_order::acquire) \|\|	145✔
242	m_queue_size.load(std::memory_order::acquire) > 0 \|\|	290✔
243	m_shutdown_requested.load(std::memory_order::acquire);	211✔
244	});
245	}	78✔
246	}
247
248	// while true; do gdb -ex run -ex quit --args taskset -c 0-1 ./test/libcoro_test "[thread_pool_ws]"; done
249
250	// Run until the queue is fully drained.
251	while (m_queue_size.load(std::memory_order::acquire) > 0)	57✔
252	{
253	(void)drain_thread_queue(thread_queue);	1✔
254	(void)try_steal(idx);	1✔
255	(void)drain_global_queue(thread_queue);	1✔
256	}
257
258	if (m_opts.on_thread_stop_functor != nullptr)	28✔
259	{
NEW 260	m_opts.on_thread_stop_functor(idx);	×
261	}
262
263	m_workers_stopped.fetch_add(1, std::memory_order::release);	28✔
264	}	28✔
265
266	auto thread_pool_ws::drain_thread_queue(riften::Deque<schedule_operation*>& queue) -> bool	870,177✔
267	{
268	bool had_task{false};	870,177✔
269	while (!queue.empty())	4,914,049✔
270	{
271	had_task \|= resume_task(queue.pop());	4,208,851✔
272	}
273	return had_task;	870,215✔
274	}
275
276	auto thread_pool_ws::try_steal(uint32_t my_idx) -> bool	870,124✔
277	{
278	bool had_tasks{false};	870,124✔
279	auto queue_size = m_workers.size();	870,124✔
280	for (size_t i = 1; i < queue_size; ++i)	918,437✔
281	{
282	auto steal_idx = (my_idx + i) % queue_size;	48,256✔
283	had_tasks \|= drain_peer_queue(m_workers[steal_idx]->m_queue);	48,256✔
284	}
285	return had_tasks;	870,181✔
286	}
287
288	auto thread_pool_ws::drain_peer_queue(riften::Deque<schedule_operation*>& queue) -> bool	35,589✔
289	{
290	bool had_task{false};	35,589✔
291	while (!queue.empty())	940,722✔
292	{
293	had_task \|= resume_task(queue.steal());	896,838✔
294	}
295	return had_task;	49,103✔
296	}
297
298	auto thread_pool_ws::drain_global_queue(riften::Deque<schedule_operation*>& queue) -> bool	870,160✔
299	{
300	if (m_global_queue.load(std::memory_order::acquire) != nullptr)	870,160✔
301	{
302	auto* head_op = detail::awaiter_list_pop_all(m_global_queue);	198,495✔
303	if (head_op != nullptr)	198,515✔
304	{
305	while (head_op != nullptr)	499,359✔
306	{
307	auto* next_op = head_op->m_next;	300,866✔
308	queue.emplace(head_op);	300,866✔
309	head_op = next_op;	300,846✔
310	}
311	}
312
313	// 1. If we got the global list we have tasks.
314	// 2. If we didn't get the global list we know we can possibly steal.
315	// Either way there should be tasks to process.
316	return true;	198,495✔
317	}
318
319	return false;	671,715✔
320	}
321
322	auto thread_pool_ws::resume_task(std::optional<schedule_operation*> op) -> bool	4,795,442✔
323	{
324	if (op.has_value())	4,795,442✔
325	{
326	m_queue_size.fetch_sub(1, std::memory_order::release);	5,097,551✔
327	auto v = op.value();	5,097,551✔
328	auto allocated = v->m_allocated.load(std::memory_order::acquire);	4,789,188✔
329	v->m_awaiting_coroutine.load(std::memory_order::acquire).resume();	4,577,782✔
330	m_size.fetch_sub(1, std::memory_order::release);	4,775,987✔
331	if (allocated)	4,775,987✔
332	{
333	delete v;	200,012✔
334	}
335	return true;	4,804,044✔
336	}
337	return false;
338	}
339
340	auto thread_pool_ws::try_wake_worker() noexcept -> void	4,744,951✔
341	{
342	// We're shutting down, no need to even attempt to wake any workers.
343	if (m_shutdown_requested.load(std::memory_order::acquire))	4,744,951✔
344	{
NEW 345	m_try_wake_workers_size.fetch_sub(1, std::memory_order::release);	×
NEW 346	return;	×
347	}
348
349	auto asleep = m_sleeping.load(std::memory_order::acquire);	4,613,692✔
350	while (asleep > 0 && !m_shutdown_requested.load(std::memory_order::acquire))	4,907,860✔
351	{
352	if (m_sleeping.compare_exchange_weak(	110✔
353	asleep, asleep - 1, std::memory_order::acq_rel, std::memory_order::acquire))
354	{
355	for (auto & info : m_workers)	119✔
356	{
357	std::unique_lock lk {info->m_wait_mutex};	112✔
358	if (info->m_sleeping.load(std::memory_order::acquire))	112✔
359	{
360	info->m_sleeping.store(false, std::memory_order::release);	48✔
361	info->m_wait_cv.notify_one();	48✔
362	break; // for	48✔
363	}
364	}	112✔
365	break; // while	55✔
366	}
367	}
368
369	m_try_wake_workers_size.fetch_sub(1, std::memory_order::release);	4,907,860✔
370
371	// // Attempt to wake a sleeper if there are any.
372	// if (m_sleeping.load(std::memory_order::acquire) > 0)
373	// {
374	// std::unique_lock<std::mutex> lk{m_wait_mutex};
375	// // Check again after acquiring the lock to see if there are any sleeping workers.
376	// if (m_sleeping.load(std::memory_order::acquire) == 0)
377	// {
378	// return;
379	// }
380
381	// m_sleeping.fetch_sub(1, std::memory_order::release);
382	// m_wait_cv.notify_one();
383	// }
384	}
385
386	} // namespace coro

jbaldwin / libcoro / 24791269108

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