14010016254

Committed 22 Mar 2025 03:38PM UTC coverage: 83.961%. First build

Build # 14010016254

Build Type

Pull #297

github

Committed by

web-flow

Commit Message

Merge 9fea175a4 into c298be096

Pull Request Pull Request #297: Add condition_variable

Run Details

131 of 195 new or added lines in 8 files covered. (67.18%)

1539 of 1833 relevant lines covered (83.96%)

3808084.26 hits per line

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

64.29

/src/condition_variable.cpp

#include "coro/condition_variable.hpp"

// this is necessary for std::lock_guard
#include <cassert>
#include <mutex>

namespace coro
{

#ifdef LIBCORO_FEATURE_NETWORKING

auto detail::strategy_based_on_io_scheduler::wait_for_ms(scoped_lock& lock, const std::chrono::milliseconds duration)
    -> task<std::cv_status>
{
    assert(!m_scheduler.expired());

    auto mtx = lock.mutex();
    lock.unlock();

    if (auto sched = m_scheduler.lock())
    {
        auto result = co_await sched->schedule(wait_task(), duration);

        auto ulock = co_await mtx->lock();
        lock       = std::move(ulock);
        co_return result.has_value() ? std::cv_status::no_timeout : std::cv_status::timeout;
    }

    co_return std::cv_status::timeout;
}

auto detail::strategy_based_on_io_scheduler::wait_task() -> task<bool>
{
    co_await wait_for_notify();
    co_return true;
}

void detail::strategy_based_on_io_scheduler::lock() noexcept
{
    while (true)
    {
        void* unlocked{};
        if (m_lock.compare_exchange_weak(unlocked, this, std::memory_order::release, std::memory_order::acquire))
        {
            break;
        }
    }
}

void detail::strategy_based_on_io_scheduler::unlock() noexcept
{
    m_lock.store(nullptr, std::memory_order::release);
}

void detail::strategy_based_on_io_scheduler::insert_waiter(wait_operation* waiter) noexcept
{
    std::lock_guard<detail::strategy_based_on_io_scheduler> guard{*this};
    m_internal_waiters.insert(waiter->m_awaiting_coroutine);
}

bool detail::strategy_based_on_io_scheduler::extract_waiter(wait_operation* waiter) noexcept
{
    std::lock_guard<detail::strategy_based_on_io_scheduler> guard{*this};
    return m_internal_waiters.erase(waiter->m_awaiting_coroutine);
}

detail::strategy_based_on_io_scheduler::strategy_based_on_io_scheduler(std::shared_ptr<io_scheduler> io_scheduler)
    : m_scheduler(io_scheduler)
{
    assert(io_scheduler);
}

void detail::strategy_based_on_io_scheduler::notify_one() noexcept
{
    assert(!m_scheduler.expired());

    if (auto waiter_guard = extract_one())
    {
        auto h = waiter_guard.value();

        if (h && !h.done())
        {
            if (auto sched = m_scheduler.lock())
            {
                sched->resume(h);
            }
        }
    }
}

void detail::strategy_based_on_io_scheduler::notify_all() noexcept
{
    assert(!m_scheduler.expired());

    if (auto waiter_guard = extract_all())
    {
        auto values = waiter_guard.values();

        if (auto sched = m_scheduler.lock())
        {
            for (const auto& h : values)
            {
                if (h && !h.done())
                {
                    sched->resume(h);
                }
            }
        }
    }
}

detail::strategy_based_on_io_scheduler::wait_operation_guard detail::strategy_based_on_io_scheduler::extract_all()
{
    wait_operation_guard result{this};
    result.set_values(m_internal_waiters);
    m_internal_waiters.clear();
    return result;
}

detail::strategy_based_on_io_scheduler::wait_operation_guard detail::strategy_based_on_io_scheduler::extract_one()
{
    wait_operation_guard result{this};
    if (!m_internal_waiters.empty())
    {
        result.set_value(*m_internal_waiters.begin());
        m_internal_waiters.erase(m_internal_waiters.begin());
    }
    return result;
}

detail::strategy_based_on_io_scheduler::wait_operation::wait_operation(detail::strategy_based_on_io_scheduler& strategy)
    : m_strategy(strategy)
{
}

detail::strategy_based_on_io_scheduler::wait_operation::~wait_operation()
{
    m_strategy.extract_waiter(this);
}

auto detail::strategy_based_on_io_scheduler::wait_operation::await_suspend(
    std::coroutine_handle<> awaiting_coroutine) noexcept -> bool
{
    m_awaiting_coroutine = awaiting_coroutine;
    m_strategy.insert_waiter(this);
    return true;
}

void detail::strategy_based_on_io_scheduler::wait_operation::await_resume() noexcept
{
}

detail::strategy_based_on_io_scheduler::wait_operation_guard::wait_operation_guard(
    detail::strategy_based_on_io_scheduler* strategy) noexcept
    : m_strategy(strategy)
{
    m_strategy->lock();
}

detail::strategy_based_on_io_scheduler::wait_operation_guard::~wait_operation_guard()
{
    m_strategy->unlock();
}

detail::strategy_based_on_io_scheduler::wait_operation_guard::operator bool() const noexcept
{
    return !m_values.empty();
}

void detail::strategy_based_on_io_scheduler::wait_operation_guard::set_value(std::coroutine_handle<> value) noexcept
{
    m_values = {value};
}

std::coroutine_handle<> detail::strategy_based_on_io_scheduler::wait_operation_guard::value() const noexcept
{
    if (m_values.empty())
        return nullptr;

    return *m_values.begin();
}

void detail::strategy_based_on_io_scheduler::wait_operation_guard::set_values(
    std::unordered_set<std::coroutine_handle<>> values) noexcept
{
    m_values = values;
}

std::unordered_set<std::coroutine_handle<>>
    detail::strategy_based_on_io_scheduler::wait_operation_guard::values() const noexcept
{
    return m_values;
}

#endif

detail::strategy_base::wait_operation::wait_operation(detail::strategy_base& strategy) : m_strategy(strategy)
{
}

bool detail::strategy_base::wait_operation::await_suspend(std::coroutine_handle<> awaiting_coroutine) noexcept
{
    m_awaiting_coroutine = awaiting_coroutine;
    while (true)
    {
        wait_operation* current = m_strategy.m_internal_waiters.load(std::memory_order::acquire);
        m_next.store(current, std::memory_order::release);

        if (!m_strategy.m_internal_waiters.compare_exchange_weak(
                current, this, std::memory_order::release, std::memory_order::acquire))
        {
            continue;
        }

        break;
    }
    return true;
}

void detail::strategy_base::wait_operation::await_resume() noexcept
{
}

void detail::strategy_base::notify_one() noexcept
{
    while (true)
    {
        auto* current = m_internal_waiters.load(std::memory_order::acquire);
        if (!current)
        {
            break;
        }

        auto* next = current->m_next.load(std::memory_order::acquire);
        if (!m_internal_waiters.compare_exchange_weak(
                current, next, std::memory_order::release, std::memory_order::acquire))
        {
            continue;
        }

        current->m_awaiting_coroutine.resume();
        break;
    }
}

void detail::strategy_base::notify_all() noexcept
{
    while (true)
    {
        auto* current = m_internal_waiters.load(std::memory_order::acquire);
        if (!current)
        {
            break;
        }

        if (!m_internal_waiters.compare_exchange_weak(
                current, nullptr, std::memory_order::release, std::memory_order::acquire))
        {
            continue;
        }

        auto* next         = current->m_next.load(std::memory_order::acquire);
        auto* locked_value = get_locked_value();

        if (next == locked_value)
        {
            // another thread in notify_all() has already taken this waiter
            break;
        }

        if (!current->m_next.compare_exchange_weak(
                next, locked_value, std::memory_order::release, std::memory_order::acquire))
        {
            continue;
        }

        current->m_awaiting_coroutine.resume();

        do
        {
            current = next;
            next    = current->m_next.load(std::memory_order::acquire);
            current->m_awaiting_coroutine.resume();
        } while (next);
    }
}

detail::strategy_base::wait_operation* detail::strategy_base::get_locked_value() noexcept
{
    return reinterpret_cast<detail::strategy_base::wait_operation*>(this);
}

auto detail::strategy_base::wait_for_notify() -> detail::strategy_base::wait_operation
{
    return wait_operation{*this};
}

} // namespace coro

1	#include "coro/condition_variable.hpp"
2
3	// this is necessary for std::lock_guard
4	#include <cassert>
5	#include <mutex>
6
7	namespace coro
8	{
9
10	#ifdef LIBCORO_FEATURE_NETWORKING
11
12	auto detail::strategy_based_on_io_scheduler::wait_for_ms(scoped_lock& lock, const std::chrono::milliseconds duration)	51✔
13	-> task<std::cv_status>
14	{
15	assert(!m_scheduler.expired());
16
17	auto mtx = lock.mutex();
18	lock.unlock();
19
20	if (auto sched = m_scheduler.lock())
21	{
22	auto result = co_await sched->schedule(wait_task(), duration);
23
24	auto ulock = co_await mtx->lock();
25	lock = std::move(ulock);
26	co_return result.has_value() ? std::cv_status::no_timeout : std::cv_status::timeout;
27	}
28
29	co_return std::cv_status::timeout;
30	}	102✔
31
32	auto detail::strategy_based_on_io_scheduler::wait_task() -> task<bool>	51✔
33	{
34	co_await wait_for_notify();
35	co_return true;
36	}	102✔
37
38	void detail::strategy_based_on_io_scheduler::lock() noexcept	70,019✔
39	{
40	while (true)
41	{
42	void* unlocked{};	70,019✔
43	if (m_lock.compare_exchange_weak(unlocked, this, std::memory_order::release, std::memory_order::acquire))	70,019✔
44	{
45	break;	12,535✔
46	}
47	}	57,486✔
48	}	12,535✔
49
50	void detail::strategy_based_on_io_scheduler::unlock() noexcept	12,537✔
51	{
52	m_lock.store(nullptr, std::memory_order::release);	12,537✔
53	}	12,537✔
54
55	void detail::strategy_based_on_io_scheduler::insert_waiter(wait_operation* waiter) noexcept	1,271✔
56	{
57	std::lock_guard<detail::strategy_based_on_io_scheduler> guard{*this};	1,271✔
58	m_internal_waiters.insert(waiter->m_awaiting_coroutine);	1,273✔
59	}	1,273✔
60
61	bool detail::strategy_based_on_io_scheduler::extract_waiter(wait_operation* waiter) noexcept	1,257✔
62	{
63	std::lock_guard<detail::strategy_based_on_io_scheduler> guard{*this};	1,257✔
64	return m_internal_waiters.erase(waiter->m_awaiting_coroutine);	1,258✔
65	}	1,258✔
66
67	detail::strategy_based_on_io_scheduler::strategy_based_on_io_scheduler(std::shared_ptr<io_scheduler> io_scheduler)	4✔
68	: m_scheduler(io_scheduler)	4✔
69	{
70	assert(io_scheduler);	4✔
71	}	4✔
72
73	void detail::strategy_based_on_io_scheduler::notify_one() noexcept	10,002✔
74	{
75	assert(!m_scheduler.expired());	10,002✔
76
77	if (auto waiter_guard = extract_one())	10,002✔
78	{
79	auto h = waiter_guard.value();	1,159✔
80
81	if (h && !h.done())	1,159✔
82	{
83	if (auto sched = m_scheduler.lock())	1,159✔
84	{
85	sched->resume(h);	1,159✔
86	}	1,159✔
87	}
88	}	10,002✔
89	}	10,002✔
90
91	void detail::strategy_based_on_io_scheduler::notify_all() noexcept	4✔
92	{
93	assert(!m_scheduler.expired());	4✔
94
95	if (auto waiter_guard = extract_all())	4✔
96	{
97	auto values = waiter_guard.values();	4✔
98
99	if (auto sched = m_scheduler.lock())	4✔
100	{
101	for (const auto& h : values)	103✔
102	{
103	if (h && !h.done())	99✔
104	{
105	sched->resume(h);	99✔
106	}
107	}
108	}	4✔
109	}	8✔
110	}	4✔
111
112	detail::strategy_based_on_io_scheduler::wait_operation_guard detail::strategy_based_on_io_scheduler::extract_all()	4✔
113	{
114	wait_operation_guard result{this};	4✔
115	result.set_values(m_internal_waiters);	4✔
116	m_internal_waiters.clear();	4✔
117	return result;	4✔
NEW 118	}	×
119
120	detail::strategy_based_on_io_scheduler::wait_operation_guard detail::strategy_based_on_io_scheduler::extract_one()	10,002✔
121	{
122	wait_operation_guard result{this};	10,002✔
123	if (!m_internal_waiters.empty())	10,002✔
124	{
125	result.set_value(*m_internal_waiters.begin());	1,159✔
126	m_internal_waiters.erase(m_internal_waiters.begin());	1,159✔
127	}
128	return result;	10,002✔
NEW 129	}	×
130
131	detail::strategy_based_on_io_scheduler::wait_operation::wait_operation(detail::strategy_based_on_io_scheduler& strategy)	1,272✔
132	: m_strategy(strategy)	1,272✔
133	{
134	}	1,272✔
135
136	detail::strategy_based_on_io_scheduler::wait_operation::~wait_operation()	1,257✔
137	{
138	m_strategy.extract_waiter(this);	1,257✔
139	}	1,258✔
140
141	auto detail::strategy_based_on_io_scheduler::wait_operation::await_suspend(	1,271✔
142	std::coroutine_handle<> awaiting_coroutine) noexcept -> bool
143	{
144	m_awaiting_coroutine = awaiting_coroutine;	1,271✔
145	m_strategy.insert_waiter(this);	1,271✔
146	return true;	1,273✔
147	}
148
149	void detail::strategy_based_on_io_scheduler::wait_operation::await_resume() noexcept	1,257✔
150	{
151	}	1,257✔
152
153	detail::strategy_based_on_io_scheduler::wait_operation_guard::wait_operation_guard(	10,006✔
154	detail::strategy_based_on_io_scheduler* strategy) noexcept	10,006✔
155	: m_strategy(strategy)	10,006✔
156	{
157	m_strategy->lock();	10,006✔
158	}	10,006✔
159
160	detail::strategy_based_on_io_scheduler::wait_operation_guard::~wait_operation_guard()	10,006✔
161	{
162	m_strategy->unlock();	10,006✔
163	}	10,006✔
164
165	detail::strategy_based_on_io_scheduler::wait_operation_guard::operator bool() const noexcept	10,006✔
166	{
167	return !m_values.empty();	10,006✔
168	}
169
170	void detail::strategy_based_on_io_scheduler::wait_operation_guard::set_value(std::coroutine_handle<> value) noexcept	1,159✔
171	{
172	m_values = {value};	1,159✔
173	}	1,159✔
174
175	std::coroutine_handle<> detail::strategy_based_on_io_scheduler::wait_operation_guard::value() const noexcept	1,159✔
176	{
177	if (m_values.empty())	1,159✔
NEW 178	return nullptr;	×
179
180	return *m_values.begin();	1,159✔
181	}
182
183	void detail::strategy_based_on_io_scheduler::wait_operation_guard::set_values(	4✔
184	std::unordered_set<std::coroutine_handle<>> values) noexcept
185	{
186	m_values = values;	4✔
187	}	4✔
188
189	std::unordered_set<std::coroutine_handle<>>
190	detail::strategy_based_on_io_scheduler::wait_operation_guard::values() const noexcept	4✔
191	{
192	return m_values;	4✔
193	}
194
195	#endif
196
NEW 197	detail::strategy_base::wait_operation::wait_operation(detail::strategy_base& strategy) : m_strategy(strategy)	×
198	{
NEW 199	}	×
200
NEW 201	bool detail::strategy_base::wait_operation::await_suspend(std::coroutine_handle<> awaiting_coroutine) noexcept	×
202	{
NEW 203	m_awaiting_coroutine = awaiting_coroutine;	×
204	while (true)
205	{
NEW 206	wait_operation* current = m_strategy.m_internal_waiters.load(std::memory_order::acquire);	×
NEW 207	m_next.store(current, std::memory_order::release);	×
208
NEW 209	if (!m_strategy.m_internal_waiters.compare_exchange_weak(	×
210	current, this, std::memory_order::release, std::memory_order::acquire))
211	{
NEW 212	continue;	×
213	}
214
NEW 215	break;	×
NEW 216	}	×
NEW 217	return true;	×
218	}
219
NEW 220	void detail::strategy_base::wait_operation::await_resume() noexcept	×
221	{
NEW 222	}	×
223
NEW 224	void detail::strategy_base::notify_one() noexcept	×
225	{
226	while (true)
227	{
NEW 228	auto* current = m_internal_waiters.load(std::memory_order::acquire);	×
NEW 229	if (!current)	×
230	{
NEW 231	break;	×
232	}
233
NEW 234	auto* next = current->m_next.load(std::memory_order::acquire);	×
NEW 235	if (!m_internal_waiters.compare_exchange_weak(	×
236	current, next, std::memory_order::release, std::memory_order::acquire))
237	{
NEW 238	continue;	×
239	}
240
NEW 241	current->m_awaiting_coroutine.resume();	×
NEW 242	break;	×
NEW 243	}	×
NEW 244	}	×
245
NEW 246	void detail::strategy_base::notify_all() noexcept	×
247	{
248	while (true)
249	{
NEW 250	auto* current = m_internal_waiters.load(std::memory_order::acquire);	×
NEW 251	if (!current)	×
252	{
NEW 253	break;	×
254	}
255
NEW 256	if (!m_internal_waiters.compare_exchange_weak(	×
257	current, nullptr, std::memory_order::release, std::memory_order::acquire))
258	{
NEW 259	continue;	×
260	}
261
NEW 262	auto* next = current->m_next.load(std::memory_order::acquire);	×
NEW 263	auto* locked_value = get_locked_value();	×
264
NEW 265	if (next == locked_value)	×
266	{
267	// another thread in notify_all() has already taken this waiter
NEW 268	break;	×
269	}
270
NEW 271	if (!current->m_next.compare_exchange_weak(	×
272	next, locked_value, std::memory_order::release, std::memory_order::acquire))
273	{
NEW 274	continue;	×
275	}
276
NEW 277	current->m_awaiting_coroutine.resume();	×
278
279	do
280	{
NEW 281	current = next;	×
NEW 282	next = current->m_next.load(std::memory_order::acquire);	×
NEW 283	current->m_awaiting_coroutine.resume();	×
NEW 284	} while (next);	×
NEW 285	}	×
NEW 286	}	×
287
NEW 288	detail::strategy_base::wait_operation* detail::strategy_base::get_locked_value() noexcept	×
289	{
NEW 290	return reinterpret_cast<detail::strategy_base::wait_operation*>(this);	×
291	}
292
NEW 293	auto detail::strategy_base::wait_for_notify() -> detail::strategy_base::wait_operation	×
294	{
NEW 295	return wait_operation{*this};	×
296	}
297
298	} // namespace coro

jbaldwin / libcoro / 14010016254

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