15356501868

Committed 30 May 2025 09:57PM UTC coverage: 87.567%. First build

Build # 15356501868

Build Type

Pull #334

github

Committed by

web-flow

Commit Message

Merge 818f341cf into ae3662d54

Pull Request Pull Request #334: Semaphore test failure

Run Details

27 of 28 new or added lines in 3 files covered. (96.43%)

1634 of 1866 relevant lines covered (87.57%)

3784856.12 hits per line

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

93.33

/src/semaphore.cpp

#include "coro/semaphore.hpp"

namespace coro
{
using namespace std::string_literals;

std::string semaphore::acquire_result_acquired          = "acquired"s;
std::string semaphore::acquire_result_semaphore_stopped = "semaphore_stopped"s;
std::string semaphore::acquire_result_unknown           = "unknown"s;

semaphore::semaphore(std::ptrdiff_t max_value_and_starting_value)
    : semaphore(max_value_and_starting_value, max_value_and_starting_value)
{
}

semaphore::semaphore(std::ptrdiff_t max_value, std::ptrdiff_t starting_value)
    : m_max_value(max_value),
      m_counter(starting_value <= max_value ? starting_value : max_value)
{
}

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

semaphore::acquire_operation::acquire_operation(semaphore& s) : m_semaphore(s)
{
}

auto semaphore::acquire_operation::await_ready() const noexcept -> bool
{
    if (m_semaphore.m_shutdown.load(std::memory_order::acquire))
    {
        return true;
    }
    return m_semaphore.try_acquire();
}

auto semaphore::acquire_operation::await_suspend(std::coroutine_handle<> awaiting_coroutine) noexcept -> bool
{
    std::unique_lock lk{m_semaphore.m_waiter_mutex};
    if (m_semaphore.m_shutdown.load(std::memory_order::relaxed))
    {
        return false;
    }

    if (m_semaphore.try_acquire())
    {
        return false;
    }

    if (m_semaphore.m_acquire_waiters == nullptr)
    {
        m_semaphore.m_acquire_waiters = this;
    }
    else
    {
        // Set our next to the current head.
        m_next = m_semaphore.m_acquire_waiters;
        // Set the semaphore head to this.
        m_semaphore.m_acquire_waiters = this;
    }

    m_awaiting_coroutine = awaiting_coroutine;
    return true;
}

auto semaphore::acquire_operation::await_resume() const -> acquire_result
{
    if (m_semaphore.m_shutdown.load(std::memory_order::acquire))
    {
        return acquire_result::semaphore_stopped;
    }
    return acquire_result::acquired;
}

auto semaphore::release() -> void
{
    // It seems like the atomic counter could be incremented, but then resuming a waiter could have
    // a race between a new acquirer grabbing the just incremented resource value from us.  So its
    // best to check if there are any waiters first, and transfer owernship of the resource thats
    // being released directly to the waiter to avoid this problem.

    std::unique_lock lk{m_waiter_mutex};
    if (m_acquire_waiters != nullptr)
    {
        acquire_operation* to_resume = m_acquire_waiters;
        m_acquire_waiters            = m_acquire_waiters->m_next;
        lk.unlock();

        // This will transfer ownership of the resource to the resumed waiter.
        to_resume->m_awaiting_coroutine.resume();
    }
    else
    {
        // Optimistically increment, if we're over the max remove it.
        if (m_counter.fetch_add(1, std::memory_order::relaxed) > m_max_value)
        {
            m_counter.fetch_sub(1, std::memory_order::relaxed);
        }
    }
}

auto semaphore::try_acquire() -> bool
{
    auto expected = m_counter.load(std::memory_order::acquire);
    do
    {
        if (expected <= 0)
        {
            return false;
        }
    } while (!m_counter.compare_exchange_weak(expected, expected - 1, std::memory_order::acq_rel, std::memory_order::acquire));

    return true;
}

auto semaphore::shutdown() noexcept -> void
{
    bool expected{false};
    if (m_shutdown.compare_exchange_strong(expected, true, std::memory_order::release, std::memory_order::relaxed))
    {
        while (true)
        {
            std::unique_lock lk{m_waiter_mutex};
            if (m_acquire_waiters != nullptr)
            {
                acquire_operation* to_resume = m_acquire_waiters;
                m_acquire_waiters            = m_acquire_waiters->m_next;
                lk.unlock();

                to_resume->m_awaiting_coroutine.resume();
            }
            else
            {
                break;
            }
        }
    }
}

} // namespace coro

1	#include "coro/semaphore.hpp"
2
3	namespace coro
4	{
5	using namespace std::string_literals;
6
7	std::string semaphore::acquire_result_acquired = "acquired"s;
8	std::string semaphore::acquire_result_semaphore_stopped = "semaphore_stopped"s;
9	std::string semaphore::acquire_result_unknown = "unknown"s;
10
11	semaphore::semaphore(std::ptrdiff_t max_value_and_starting_value)	2✔
12	: semaphore(max_value_and_starting_value, max_value_and_starting_value)	2✔
13	{
14	}	2✔
15
16	semaphore::semaphore(std::ptrdiff_t max_value, std::ptrdiff_t starting_value)	4✔
17	: m_max_value(max_value),	4✔
18	m_counter(starting_value <= max_value ? starting_value : max_value)	4✔
19	{
20	}	4✔
21
22	semaphore::~semaphore()	4✔
23	{
24	shutdown();	4✔
25	}	4✔
26
27	semaphore::acquire_operation::acquire_operation(semaphore& s) : m_semaphore(s)	100,019✔
28	{
29	}	100,015✔
30
31	auto semaphore::acquire_operation::await_ready() const noexcept -> bool	100,005✔
32	{
33	if (m_semaphore.m_shutdown.load(std::memory_order::acquire))	100,005✔
34	{
35	return true;	×
36	}
37	return m_semaphore.try_acquire();	100,010✔
38	}
39
40	auto semaphore::acquire_operation::await_suspend(std::coroutine_handle<> awaiting_coroutine) noexcept -> bool	99,990✔
41	{
42	std::unique_lock lk{m_semaphore.m_waiter_mutex};	99,990✔
43	if (m_semaphore.m_shutdown.load(std::memory_order::relaxed))	100,028✔
44	{
45	return false;	×
46	}
47
48	if (m_semaphore.try_acquire())	100,028✔
49	{
50	return false;	×
51	}
52
53	if (m_semaphore.m_acquire_waiters == nullptr)	100,028✔
54	{
55	m_semaphore.m_acquire_waiters = this;	10✔
56	}
57	else
58	{
59	// Set our next to the current head.
60	m_next = m_semaphore.m_acquire_waiters;	100,018✔
61	// Set the semaphore head to this.
62	m_semaphore.m_acquire_waiters = this;	100,018✔
63	}
64
65	m_awaiting_coroutine = awaiting_coroutine;	100,028✔
66	return true;	100,028✔
67	}	100,028✔
68
69	auto semaphore::acquire_operation::await_resume() const -> acquire_result	100,033✔
70	{
71	if (m_semaphore.m_shutdown.load(std::memory_order::acquire))	100,033✔
72	{
73	return acquire_result::semaphore_stopped;	17✔
74	}
75	return acquire_result::acquired;	100,016✔
76	}
77
78	auto semaphore::release() -> void	100,016✔
79	{
80	// It seems like the atomic counter could be incremented, but then resuming a waiter could have
81	// a race between a new acquirer grabbing the just incremented resource value from us. So its
82	// best to check if there are any waiters first, and transfer owernship of the resource thats
83	// being released directly to the waiter to avoid this problem.
84
85	std::unique_lock lk{m_waiter_mutex};	100,016✔
86	if (m_acquire_waiters != nullptr)	100,016✔
87	{
88	acquire_operation* to_resume = m_acquire_waiters;	100,011✔
89	m_acquire_waiters = m_acquire_waiters->m_next;	100,011✔
90	lk.unlock();	100,011✔
91
92	// This will transfer ownership of the resource to the resumed waiter.
93	to_resume->m_awaiting_coroutine.resume();	100,011✔
94	}
95	else
96	{
97	// Optimistically increment, if we're over the max remove it.
98	if (m_counter.fetch_add(1, std::memory_order::relaxed) > m_max_value)	10✔
99	{
NEW 100	m_counter.fetch_sub(1, std::memory_order::relaxed);	×
101	}
102	}
103	}	100,016✔
104
105	auto semaphore::try_acquire() -> bool	200,000✔
106	{
107	auto expected = m_counter.load(std::memory_order::acquire);	400,001✔
108	do
109	{
110	if (expected <= 0)	200,016✔
111	{
112	return false;	200,009✔
113	}
114	} while (!m_counter.compare_exchange_weak(expected, expected - 1, std::memory_order::acq_rel, std::memory_order::acquire));	14✔
115
116	return true;
117	}
118
119	auto semaphore::shutdown() noexcept -> void	6✔
120	{
121	bool expected{false};	6✔
122	if (m_shutdown.compare_exchange_strong(expected, true, std::memory_order::release, std::memory_order::relaxed))	6✔
123	{
124	while (true)
125	{
126	std::unique_lock lk{m_waiter_mutex};	21✔
127	if (m_acquire_waiters != nullptr)	21✔
128	{
129	acquire_operation* to_resume = m_acquire_waiters;	17✔
130	m_acquire_waiters = m_acquire_waiters->m_next;	17✔
131	lk.unlock();	17✔
132
133	to_resume->m_awaiting_coroutine.resume();	17✔
134	}
135	else
136	{
137	break;	4✔
138	}
139	}	38✔
140	}
141	}	6✔
142
143	} // namespace coro

jbaldwin / libcoro / 15356501868

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