13591320473

Committed 28 Feb 2025 03:31PM UTC coverage: 86.479%. First build

Build # 13591320473

Build Type

Pull #297

github

Committed by

web-flow

Commit Message

Merge 6f84a94a6 into c298be096

Pull Request Pull Request #297: Add condition_variable

Run Details

120 of 137 new or added lines in 6 files covered. (87.59%)

1535 of 1775 relevant lines covered (86.48%)

3936978.79 hits per line

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

85.59

/src/condition_variable.cpp

#include "coro/condition_variable.hpp"

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

namespace coro
{

condition_variable::condition_variable(std::shared_ptr<io_scheduler> scheduler) : m_scheduler(scheduler)
{
}

void condition_variable::notify_one() noexcept
{
    assert(m_scheduler);

    if (auto waiter_guard = extract_one())
    {
        m_scheduler->resume(waiter_guard.value()->m_awaiting_coroutine);
    }
}

void condition_variable::notify_all() noexcept
{
    assert(m_scheduler);

    if (auto waiter_guard = extract_all())
    {
        auto* waiter = waiter_guard.value();
        do
        {
            auto* next = waiter->m_next.load(std::memory_order::acquire);
            m_scheduler->resume(waiter->m_awaiting_coroutine);
            waiter = next;
        } while (waiter);
    }
}

auto condition_variable::wait(scoped_lock& lock) -> task<void>
{
    using namespace std::chrono_literals;

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

    co_await wait_for_notify();

    auto ulock = co_await mtx->lock();
    lock       = std::move(ulock);
    co_return;
}

std::shared_ptr<io_scheduler> condition_variable::scheduler() const noexcept
{
    return m_scheduler;
}

void condition_variable::set_scheduler(std::shared_ptr<io_scheduler> scheduler)
{
    m_scheduler = scheduler;
}

auto condition_variable::wait_for_ms(scoped_lock& lock, const std::chrono::milliseconds duration)
    -> task<std::cv_status>
{
    assert(m_scheduler);

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

    auto result = co_await m_scheduler->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;
}

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

void condition_variable::lock() noexcept
{
    while (true)
    {
        void* unlocked{};
        if (m_lock.compare_exchange_weak(unlocked, this, std::memory_order::acq_rel))
        {
            break;
        }
    }
}

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

void condition_variable::insert_waiter(wait_operation* waiter) noexcept
{
    while (true)
    {
        wait_operation* current = m_internal_waiters.load(std::memory_order::acquire);
        waiter->m_next.store(current, std::memory_order::release);

        if (!m_internal_waiters.compare_exchange_weak(current, waiter, std::memory_order::acq_rel))
        {
            continue;
        }

        break;
    }
}

bool condition_variable::extract_waiter(wait_operation* waiter) noexcept
{
    std::lock_guard<condition_variable> guard{*this};
    bool                                result{};

    while (true)
    {
        wait_operation* current = m_internal_waiters.load(std::memory_order::acquire);

        if (!current)
        {
            break;
        }

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

        if (current == waiter)
        {
            if (!m_internal_waiters.compare_exchange_weak(current, next, std::memory_order::acq_rel))
            {
                continue;
            }
        }

        while (next && next != waiter)
        {
            current = next;
            next    = current->m_next.load(std::memory_order::acquire);
        }

        if (!next)
        {
            break;
        }

        wait_operation* new_next = waiter->m_next.load(std::memory_order::acquire);

        if (!current->m_next.compare_exchange_strong(next, new_next, std::memory_order::acq_rel))
        {
            continue;
        }

        waiter->m_next.store(nullptr, std::memory_order::release);
        result = true;
        break;
    }

    return result;
}

condition_variable::wait_operation_guard condition_variable::extract_all()
{
    wait_operation_guard result{this};

    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::acq_rel))
        {
            continue;
        }

        result.set_value(current);
        break;
    }

    return result;
}

condition_variable::wait_operation_guard condition_variable::extract_one()
{
    wait_operation_guard result{this};

    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::acq_rel))
        {
            continue;
        }

        current->m_next.store(nullptr, std::memory_order::release);
        result.set_value(current);
        break;
    }

    return result;
}

condition_variable::wait_operation::wait_operation(condition_variable& cv) : m_condition_variable(cv)
{
}

condition_variable::wait_operation::~wait_operation()
{
    m_condition_variable.extract_waiter(this);
}

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

void condition_variable::wait_operation::await_resume() noexcept
{
}

condition_variable::wait_operation_guard::wait_operation_guard(condition_variable* cv) noexcept : m_cv(cv)
{
    m_cv->lock();
}

condition_variable::wait_operation_guard::~wait_operation_guard()
{
    m_cv->unlock();
}

condition_variable::wait_operation_guard::operator bool() const noexcept
{
    return m_value;
}

void condition_variable::wait_operation_guard::set_value(wait_operation* value) noexcept
{
    m_value = value;
}

condition_variable::wait_operation* condition_variable::wait_operation_guard::value() const noexcept
{
    return m_value;
}

} // 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	condition_variable::condition_variable(std::shared_ptr<io_scheduler> scheduler) : m_scheduler(scheduler)	4✔
11	{
12	}	4✔
13
14	void condition_variable::notify_one() noexcept	10,002✔
15	{
16	assert(m_scheduler);	10,002✔
17
18	if (auto waiter_guard = extract_one())	10,002✔
19	{
20	m_scheduler->resume(waiter_guard.value()->m_awaiting_coroutine);	9,983✔
21	}	10,002✔
22	}	10,002✔
23
24	void condition_variable::notify_all() noexcept	4✔
25	{
26	assert(m_scheduler);	4✔
27
28	if (auto waiter_guard = extract_all())	4✔
29	{
30	auto* waiter = waiter_guard.value();	4✔
31	do
32	{
33	auto* next = waiter->m_next.load(std::memory_order::acquire);	99✔
34	m_scheduler->resume(waiter->m_awaiting_coroutine);	99✔
35	waiter = next;	99✔
36	} while (waiter);	99✔
37	}	4✔
38	}	4✔
39
40	auto condition_variable::wait(scoped_lock& lock) -> task<void>	10,046✔
41	{
42	using namespace std::chrono_literals;
43
44	auto mtx = lock.mutex();
45	lock.unlock();
46
47	co_await wait_for_notify();
48
49	auto ulock = co_await mtx->lock();
50	lock = std::move(ulock);
51	co_return;
52	}	20,092✔
53
NEW 54	std::shared_ptr<io_scheduler> condition_variable::scheduler() const noexcept	×
55	{
NEW 56	return m_scheduler;	×
57	}
58
NEW 59	void condition_variable::set_scheduler(std::shared_ptr<io_scheduler> scheduler)	×
60	{
NEW 61	m_scheduler = scheduler;	×
NEW 62	}	×
63
64	auto condition_variable::wait_for_ms(scoped_lock& lock, const std::chrono::milliseconds duration)	51✔
65	-> task<std::cv_status>
66	{
67	assert(m_scheduler);
68
69	auto mtx = lock.mutex();
70	lock.unlock();
71
72	auto result = co_await m_scheduler->schedule(wait_task(), duration);
73
74	auto ulock = co_await mtx->lock();
75	lock = std::move(ulock);
76	co_return result.has_value() ? std::cv_status::no_timeout : std::cv_status::timeout;
77	}	102✔
78
79	auto condition_variable::wait_task() -> task<bool>	51✔
80	{
81	co_await wait_for_notify();
82	co_return true;
83	}	102✔
84
85	void condition_variable::lock() noexcept	112,775✔
86	{
87	while (true)
88	{
89	void* unlocked{};	112,775✔
90	if (m_lock.compare_exchange_weak(unlocked, this, std::memory_order::acq_rel))	112,775✔
91	{
92	break;	19,935✔
93	}
94	}	92,844✔
95	}	19,935✔
96
97	void condition_variable::unlock() noexcept	20,088✔
98	{
99	m_lock.store(nullptr, std::memory_order::release);	20,088✔
100	}	20,088✔
101
102	void condition_variable::insert_waiter(wait_operation* waiter) noexcept	10,149✔
103	{
104	while (true)
105	{
106	wait_operation* current = m_internal_waiters.load(std::memory_order::acquire);	10,149✔
107	waiter->m_next.store(current, std::memory_order::release);	10,149✔
108
109	if (!m_internal_waiters.compare_exchange_weak(current, waiter, std::memory_order::acq_rel))	10,149✔
110	{
111	continue;	53✔
112	}
113
114	break;	10,097✔
115	}	53✔
116	}	10,097✔
117
118	bool condition_variable::extract_waiter(wait_operation* waiter) noexcept	10,077✔
119	{
120	std::lock_guard<condition_variable> guard{*this};	10,077✔
121	bool result{};	10,082✔
122
123	while (true)
124	{
125	wait_operation* current = m_internal_waiters.load(std::memory_order::acquire);	10,082✔
126
127	if (!current)	10,082✔
128	{
129	break;	112✔
130	}
131
132	wait_operation* next = current->m_next.load(std::memory_order::acquire);	9,970✔
133
134	if (current == waiter)	9,970✔
135	{
NEW 136	if (!m_internal_waiters.compare_exchange_weak(current, next, std::memory_order::acq_rel))	×
137	{
NEW 138	continue;	×
139	}
140	}
141
142	while (next && next != waiter)	209,646✔
143	{
144	current = next;	199,676✔
145	next = current->m_next.load(std::memory_order::acquire);	199,676✔
146	}
147
148	if (!next)	9,970✔
149	{
150	break;	9,970✔
151	}
152
NEW 153	wait_operation* new_next = waiter->m_next.load(std::memory_order::acquire);	×
154
NEW 155	if (!current->m_next.compare_exchange_strong(next, new_next, std::memory_order::acq_rel))	×
156	{
NEW 157	continue;	×
158	}
159
NEW 160	waiter->m_next.store(nullptr, std::memory_order::release);	×
NEW 161	result = true;	×
NEW 162	break;	×
NEW 163	}	×
164
165	return result;	20,164✔
166	}	10,082✔
167
168	condition_variable::wait_operation_guard condition_variable::extract_all()	4✔
169	{
170	wait_operation_guard result{this};	4✔
171
172	while (true)
173	{
174	auto* current = m_internal_waiters.load(std::memory_order::acquire);	4✔
175	if (!current)	4✔
176	{
NEW 177	break;	×
178	}
179
180	if (!m_internal_waiters.compare_exchange_weak(current, nullptr, std::memory_order::acq_rel))	4✔
181	{
NEW 182	continue;	×
183	}
184
185	result.set_value(current);	4✔
186	break;	4✔
NEW 187	}	×
188
189	return result;	4✔
190	}
191
192	condition_variable::wait_operation_guard condition_variable::extract_one()	10,002✔
193	{
194	wait_operation_guard result{this};	10,002✔
195
196	while (true)
197	{
198	auto* current = m_internal_waiters.load(std::memory_order::acquire);	10,026✔
199	if (!current)	10,026✔
200	{
201	break;	19✔
202	}
203
204	auto* next = current->m_next.load(std::memory_order::acquire);	10,007✔
205	if (!m_internal_waiters.compare_exchange_weak(current, next, std::memory_order::acq_rel))	10,007✔
206	{
207	continue;	24✔
208	}
209
210	current->m_next.store(nullptr, std::memory_order::release);	9,983✔
211	result.set_value(current);	9,983✔
212	break;	9,983✔
213	}	24✔
214
215	return result;	10,002✔
216	}
217
218	condition_variable::wait_operation::wait_operation(condition_variable& cv) : m_condition_variable(cv)	10,095✔
219	{
220	}	10,096✔
221
222	condition_variable::wait_operation::~wait_operation()	10,078✔
223	{
224	m_condition_variable.extract_waiter(this);	10,078✔
225	}	10,081✔
226
227	auto condition_variable::wait_operation::await_suspend(std::coroutine_handle<> awaiting_coroutine) noexcept -> bool	10,096✔
228	{
229	m_awaiting_coroutine = awaiting_coroutine;	10,096✔
230	m_condition_variable.insert_waiter(this);	10,096✔
231	return true;	10,097✔
232	}
233
234	void condition_variable::wait_operation::await_resume() noexcept	10,078✔
235	{
236	}	10,078✔
237
238	condition_variable::wait_operation_guard::wait_operation_guard(condition_variable* cv) noexcept : m_cv(cv)	10,006✔
239	{
240	m_cv->lock();	10,006✔
241	}	10,006✔
242
243	condition_variable::wait_operation_guard::~wait_operation_guard()	10,006✔
244	{
245	m_cv->unlock();	10,006✔
246	}	10,006✔
247
248	condition_variable::wait_operation_guard::operator bool() const noexcept	10,006✔
249	{
250	return m_value;	10,006✔
251	}
252
253	void condition_variable::wait_operation_guard::set_value(wait_operation* value) noexcept	9,987✔
254	{
255	m_value = value;	9,987✔
256	}	9,987✔
257
258	condition_variable::wait_operation* condition_variable::wait_operation_guard::value() const noexcept	9,987✔
259	{
260	return m_value;	9,987✔
261	}
262
263	} // namespace coro

jbaldwin / libcoro / 13591320473

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