14957747050

Committed 11 May 2025 04:41PM UTC coverage: 86.903%. First build

Build # 14957747050

Build Type

Pull #317

github

Committed by

web-flow

Commit Message

Merge 2c2845262 into 131cf5dca

Pull Request Pull Request #317: feat: add kqueue support

Run Details

92 of 100 new or added lines in 7 files covered. (92.0%)

1493 of 1718 relevant lines covered (86.9%)

5142782.05 hits per line

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

88.52

/src/io_notifier.cpp

#include "coro/io_notifier.hpp"

#include <array>
#include <stdexcept>

namespace coro
{

io_notifier::io_notifier()
#if defined(__FreeBSD__) || defined(__APPLE__) || defined(__OpenBSD__) || defined(__NetBSD__)
    : m_fd{::kqueue()}
#elif defined(__linux__)
    : m_fd{::epoll_create1(EPOLL_CLOEXEC)}
#endif
{
}

io_notifier::~io_notifier()
{
    if (m_fd != -1)
    {
        ::close(m_fd);
        m_fd = -1;
    }
}

timer_handle::timer_handle(const void* timer_handle_ptr, io_notifier& notifier)
#if defined(__FreeBSD__) || defined(__APPLE__) || defined(__OpenBSD__) || defined(__NetBSD__)
    : m_fd{1},
#elif defined(__linux__)
    : m_fd(::timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK | TFD_CLOEXEC)),
#endif
      m_timer_handle_ptr(timer_handle_ptr)
{
#if defined(__FreeBSD__) || defined(__APPLE__) || defined(__OpenBSD__) || defined(__NetBSD__)
    (void)notifier;
#elif defined(__linux__)
    notifier.watch(m_fd, coro::poll_op::read, const_cast<void*>(m_timer_handle_ptr), true);
#endif
}

auto io_notifier::watch_timer(const timer_handle& timer, std::chrono::nanoseconds duration) -> bool
{
#if defined(__FreeBSD__) || defined(__APPLE__) || defined(__OpenBSD__) || defined(__NetBSD__)
    auto event_data = event_t{};
    EV_SET(
        &event_data,
        timer.m_fd,
        EVFILT_TIMER,
        EV_ADD | EV_ONESHOT,
        NOTE_NSECONDS,
        duration.count(),
        const_cast<void*>(timer.m_timer_handle_ptr));
    return ::kevent(m_fd, &event_data, 1, nullptr, 0, nullptr) != -1;
#elif defined(__linux__)
    auto seconds = std::chrono::duration_cast<std::chrono::seconds>(duration);
    duration -= seconds;
    auto nanoseconds = std::chrono::duration_cast<std::chrono::nanoseconds>(duration);

    // As a safeguard if both values end up as zero (or negative) then trigger the timeout
    // immediately as zero disarms timerfd according to the man pages and negative values
    // will result in an error return value.
    if (seconds <= 0s)
    {
        seconds = 0s;
        if (nanoseconds <= 0ns)
        {
            nanoseconds = 1ns;
        }
    }

    itimerspec ts{};
    ts.it_value.tv_sec  = seconds.count();
    ts.it_value.tv_nsec = nanoseconds.count();
    return ::timerfd_settime(timer.m_fd, 0, &ts, nullptr) != -1;
#endif
}

auto io_notifier::watch(fd_t fd, coro::poll_op op, void* data, bool keep) -> bool
{
#if defined(__FreeBSD__) || defined(__APPLE__) || defined(__OpenBSD__) || defined(__NetBSD__)
    auto event_data = event_t{};
    auto mode       = EV_ADD | EV_ENABLE;
    if (!keep)
    {
        mode |= EV_ONESHOT;
    }
    EV_SET(&event_data, fd, static_cast<int16_t>(op), mode, 0, 0, data);
    return ::kevent(m_fd, &event_data, 1, nullptr, 0, nullptr) != -1;
#elif defined(__linux__)
    auto event_data     = event_t{};
    event_data.events   = static_cast<uint32_t>(op) | EPOLLRDHUP;
    event_data.data.ptr = data;
    if (!keep)
    {
        event_data.events |= EPOLLONESHOT;
    }
    return ::epoll_ctl(m_fd, EPOLL_CTL_ADD, fd, &event_data) != -1;
#endif
}

auto io_notifier::watch(detail::poll_info& pi) -> bool
{
#if defined(__FreeBSD__) || defined(__APPLE__) || defined(__OpenBSD__) || defined(__NetBSD__)
    // For read-write event, we need to register both event types separately to the kqueue
    if (pi.m_op == coro::poll_op::read_write)
    {
        auto event_data = event_t{};

        EV_SET(
            &event_data,
            pi.m_fd,
            static_cast<int16_t>(coro::poll_op::read),
            EV_ADD | EV_ONESHOT | EV_ENABLE,
            0,
            0,
            static_cast<void*>(&pi));
        ::kevent(m_fd, &event_data, 1, nullptr, 0, nullptr);

        EV_SET(
            &event_data,
            pi.m_fd,
            static_cast<int16_t>(coro::poll_op::write),
            EV_ADD | EV_ONESHOT | EV_ENABLE,
            0,
            0,
            static_cast<void*>(&pi));
        ::kevent(m_fd, &event_data, 1, nullptr, 0, nullptr);

        return true;
    }
    else
    {
        auto event_data = event_t{};
        EV_SET(
            &event_data,
            pi.m_fd,
            static_cast<int16_t>(pi.m_op),
            EV_ADD | EV_ONESHOT | EV_ENABLE,
            0,
            0,
            static_cast<void*>(&pi));
        return ::kevent(m_fd, &event_data, 1, nullptr, 0, nullptr) != -1;
    }
#elif defined(__linux__)
    auto event_data     = event_t{};
    event_data.events   = static_cast<uint32_t>(pi.m_op) | EPOLLONESHOT | EPOLLRDHUP;
    event_data.data.ptr = static_cast<void*>(&pi);
    return ::epoll_ctl(m_fd, EPOLL_CTL_ADD, pi.m_fd, &event_data) != -1;
#endif
}

auto io_notifier::unwatch(detail::poll_info& pi) -> bool
{
#if defined(__FreeBSD__) || defined(__APPLE__) || defined(__OpenBSD__) || defined(__NetBSD__)
    // For read-write event, we need to register both event types separately to the kqueue
    if (pi.m_op == coro::poll_op::read_write)
    {
        auto event_data = event_t{};

        EV_SET(&event_data, pi.m_fd, static_cast<int16_t>(coro::poll_op::read), EV_DELETE, 0, 0, nullptr);
        ::kevent(m_fd, &event_data, 1, nullptr, 0, nullptr);

        EV_SET(&event_data, pi.m_fd, static_cast<int16_t>(coro::poll_op::write), EV_DELETE, 0, 0, nullptr);
        ::kevent(m_fd, &event_data, 1, nullptr, 0, nullptr);

        return true;
    }
    else
    {
        auto event_data = event_t{};
        EV_SET(&event_data, pi.m_fd, static_cast<int16_t>(pi.m_op), EV_DELETE, 0, 0, nullptr);
        return ::kevent(m_fd, &event_data, 1, nullptr, 0, nullptr) != -1;
    }
#elif defined(__linux__)
    return ::epoll_ctl(m_fd, EPOLL_CTL_DEL, pi.m_fd, nullptr) != -1;
#endif
}

auto io_notifier::unwatch_timer(const timer_handle& timer) -> bool
{
#if defined(__FreeBSD__) || defined(__APPLE__) || defined(__OpenBSD__) || defined(__NetBSD__)
    auto event_data = event_t{};
    EV_SET(&event_data, timer.m_fd, EVFILT_TIMER, EV_DELETE, 0, 0, nullptr);
    return ::kevent(m_fd, &event_data, 1, nullptr, 0, nullptr) != -1;
#elif defined(__linux__)
    // Setting these values to zero disables the timer.
    itimerspec ts{};
    ts.it_value.tv_sec  = 0;
    ts.it_value.tv_nsec = 0;
    return ::timerfd_settime(timer.m_fd, 0, &ts, nullptr) != -1;
#endif
}

auto io_notifier::next_events(std::chrono::milliseconds timeout)
    -> std::vector<std::pair<detail::poll_info*, coro::poll_status>>
{
    auto ready_events = std::vector<std::pair<detail::poll_info*, coro::poll_status>>();
    auto ready_set    = std::array<event_t, m_max_events>{};
#if defined(__FreeBSD__) || defined(__APPLE__) || defined(__OpenBSD__) || defined(__NetBSD__)
    const auto timeout_as_secs = std::chrono::duration_cast<std::chrono::seconds>(timeout);
    auto       timeout_spec    = ::timespec{
                 .tv_sec  = timeout_as_secs.count(),
                 .tv_nsec = std::chrono::duration_cast<std::chrono::nanoseconds>(timeout - timeout_as_secs).count(),
    };
    const std::size_t num_ready = ::kevent(m_fd, nullptr, 0, ready_set.data(), ready_set.size(), &timeout_spec);
    for (std::size_t i = 0; i < num_ready; i++)
    {
        ready_events.emplace_back(
            static_cast<detail::poll_info*>(ready_set[i].udata), io_notifier::event_to_poll_status(ready_set[i]));
    }
#elif defined(__linux__)
    int num_ready = ::epoll_wait(m_fd, ready_set.data(), ready_set.size(), timeout.count());
    for (int i = 0; i < num_ready; ++i)
    {
        ready_events.emplace_back(
            static_cast<detail::poll_info*>(ready_set[i].data.ptr), io_notifier::event_to_poll_status(ready_set[i]));
    }
#endif
    return ready_events;
}

auto io_notifier::event_to_poll_status(const event_t& event) -> poll_status
{
#if defined(__FreeBSD__) || defined(__APPLE__) || defined(__OpenBSD__) || defined(__NetBSD__)
    if (event.filter & EVFILT_READ || event.filter & EVFILT_WRITE)
    {
        return poll_status::event;
    }
    else if (event.flags & EV_ERROR)
    {
        return poll_status::error;
    }
    else if (event.flags & EV_EOF)
    {
        return poll_status::closed;
    }
    throw std::runtime_error{"invalid kqueue state"};
#elif defined(__linux__)
    if (event.events & static_cast<uint32_t>(poll_op::read) || event.events & static_cast<uint32_t>(poll_op::write))
    {
        return poll_status::event;
    }
    else if (event.events & EPOLLERR)
    {
        return poll_status::error;
    }
    else if (event.events & EPOLLRDHUP || event.events & EPOLLHUP)
    {
        return poll_status::closed;
    }
    throw std::runtime_error{"invalid epoll state"};
#endif
}

} // namespace coro

1	#include "coro/io_notifier.hpp"
2
3	#include <array>
4	#include <stdexcept>
5
6	namespace coro
7	{
8
9	io_notifier::io_notifier()	43✔
10	#if defined(__FreeBSD__) \|\| defined(__APPLE__) \|\| defined(__OpenBSD__) \|\| defined(__NetBSD__)
11	: m_fd{::kqueue()}
12	#elif defined(__linux__)
13	: m_fd{::epoll_create1(EPOLL_CLOEXEC)}	43✔
14	#endif
15	{
16	}	43✔
17
18	io_notifier::~io_notifier()	22✔
19	{
20	if (m_fd != -1)	22✔
21	{
22	::close(m_fd);	22✔
23	m_fd = -1;	22✔
24	}
25	}	22✔
26
27	timer_handle::timer_handle(const void* timer_handle_ptr, io_notifier& notifier)	43✔
28	#if defined(__FreeBSD__) \|\| defined(__APPLE__) \|\| defined(__OpenBSD__) \|\| defined(__NetBSD__)
29	: m_fd{1},
30	#elif defined(__linux__)
31	: m_fd(::timerfd_create(CLOCK_MONOTONIC, TFD_NONBLOCK \| TFD_CLOEXEC)),	43✔
32	#endif
33	m_timer_handle_ptr(timer_handle_ptr)	43✔
34	{
35	#if defined(__FreeBSD__) \|\| defined(__APPLE__) \|\| defined(__OpenBSD__) \|\| defined(__NetBSD__)
36	(void)notifier;
37	#elif defined(__linux__)
38	notifier.watch(m_fd, coro::poll_op::read, const_cast<void*>(m_timer_handle_ptr), true);	43✔
39	#endif
40	}	43✔
41
42	auto io_notifier::watch_timer(const timer_handle& timer, std::chrono::nanoseconds duration) -> bool	941,692✔
43	{
44	#if defined(__FreeBSD__) \|\| defined(__APPLE__) \|\| defined(__OpenBSD__) \|\| defined(__NetBSD__)
45	auto event_data = event_t{};
46	EV_SET(
47	&event_data,
48	timer.m_fd,
49	EVFILT_TIMER,
50	EV_ADD \| EV_ONESHOT,
51	NOTE_NSECONDS,
52	duration.count(),
53	const_cast<void*>(timer.m_timer_handle_ptr));
54	return ::kevent(m_fd, &event_data, 1, nullptr, 0, nullptr) != -1;
55	#elif defined(__linux__)
56	auto seconds = std::chrono::duration_cast<std::chrono::seconds>(duration);	941,692✔
57	duration -= seconds;	941,692✔
58	auto nanoseconds = std::chrono::duration_cast<std::chrono::nanoseconds>(duration);	941,692✔
59
60	// As a safeguard if both values end up as zero (or negative) then trigger the timeout
61	// immediately as zero disarms timerfd according to the man pages and negative values
62	// will result in an error return value.
63	if (seconds <= 0s)	941,692✔
64	{
65	seconds = 0s;	941,687✔
66	if (nanoseconds <= 0ns)	941,687✔
67	{
68	nanoseconds = 1ns;	628,765✔
69	}
70	}
71
72	itimerspec ts{};	941,692✔
73	ts.it_value.tv_sec = seconds.count();	941,692✔
74	ts.it_value.tv_nsec = nanoseconds.count();	941,692✔
75	return ::timerfd_settime(timer.m_fd, 0, &ts, nullptr) != -1;	941,692✔
76	#endif
77	}
78
79	auto io_notifier::watch(fd_t fd, coro::poll_op op, void* data, bool keep) -> bool	129✔
80	{
81	#if defined(__FreeBSD__) \|\| defined(__APPLE__) \|\| defined(__OpenBSD__) \|\| defined(__NetBSD__)
82	auto event_data = event_t{};
83	auto mode = EV_ADD \| EV_ENABLE;
84	if (!keep)
85	{
86	mode \|= EV_ONESHOT;
87	}
88	EV_SET(&event_data, fd, static_cast<int16_t>(op), mode, 0, 0, data);
89	return ::kevent(m_fd, &event_data, 1, nullptr, 0, nullptr) != -1;
90	#elif defined(__linux__)
91	auto event_data = event_t{};	129✔
92	event_data.events = static_cast<uint32_t>(op) \| EPOLLRDHUP;	129✔
93	event_data.data.ptr = data;	129✔
94	if (!keep)	129✔
95	{
NEW 96	event_data.events \|= EPOLLONESHOT;	×
97	}
98	return ::epoll_ctl(m_fd, EPOLL_CTL_ADD, fd, &event_data) != -1;	129✔
99	#endif
100	}
101
102	auto io_notifier::watch(detail::poll_info& pi) -> bool	186,707✔
103	{
104	#if defined(__FreeBSD__) \|\| defined(__APPLE__) \|\| defined(__OpenBSD__) \|\| defined(__NetBSD__)
105	// For read-write event, we need to register both event types separately to the kqueue
106	if (pi.m_op == coro::poll_op::read_write)
107	{
108	auto event_data = event_t{};
109
110	EV_SET(
111	&event_data,
112	pi.m_fd,
113	static_cast<int16_t>(coro::poll_op::read),
114	EV_ADD \| EV_ONESHOT \| EV_ENABLE,
115	0,
116	0,
117	static_cast<void*>(&pi));
118	::kevent(m_fd, &event_data, 1, nullptr, 0, nullptr);
119
120	EV_SET(
121	&event_data,
122	pi.m_fd,
123	static_cast<int16_t>(coro::poll_op::write),
124	EV_ADD \| EV_ONESHOT \| EV_ENABLE,
125	0,
126	0,
127	static_cast<void*>(&pi));
128	::kevent(m_fd, &event_data, 1, nullptr, 0, nullptr);
129
130	return true;
131	}
132	else
133	{
134	auto event_data = event_t{};
135	EV_SET(
136	&event_data,
137	pi.m_fd,
138	static_cast<int16_t>(pi.m_op),
139	EV_ADD \| EV_ONESHOT \| EV_ENABLE,
140	0,
141	0,
142	static_cast<void*>(&pi));
143	return ::kevent(m_fd, &event_data, 1, nullptr, 0, nullptr) != -1;
144	}
145	#elif defined(__linux__)
146	auto event_data = event_t{};	186,707✔
147	event_data.events = static_cast<uint32_t>(pi.m_op) \| EPOLLONESHOT \| EPOLLRDHUP;	186,707✔
148	event_data.data.ptr = static_cast<void*>(&pi);	186,707✔
149	return ::epoll_ctl(m_fd, EPOLL_CTL_ADD, pi.m_fd, &event_data) != -1;	186,707✔
150	#endif
151	}
152
153	auto io_notifier::unwatch(detail::poll_info& pi) -> bool	200,065✔
154	{
155	#if defined(__FreeBSD__) \|\| defined(__APPLE__) \|\| defined(__OpenBSD__) \|\| defined(__NetBSD__)
156	// For read-write event, we need to register both event types separately to the kqueue
157	if (pi.m_op == coro::poll_op::read_write)
158	{
159	auto event_data = event_t{};
160
161	EV_SET(&event_data, pi.m_fd, static_cast<int16_t>(coro::poll_op::read), EV_DELETE, 0, 0, nullptr);
162	::kevent(m_fd, &event_data, 1, nullptr, 0, nullptr);
163
164	EV_SET(&event_data, pi.m_fd, static_cast<int16_t>(coro::poll_op::write), EV_DELETE, 0, 0, nullptr);
165	::kevent(m_fd, &event_data, 1, nullptr, 0, nullptr);
166
167	return true;
168	}
169	else
170	{
171	auto event_data = event_t{};
172	EV_SET(&event_data, pi.m_fd, static_cast<int16_t>(pi.m_op), EV_DELETE, 0, 0, nullptr);
173	return ::kevent(m_fd, &event_data, 1, nullptr, 0, nullptr) != -1;
174	}
175	#elif defined(__linux__)
176	return ::epoll_ctl(m_fd, EPOLL_CTL_DEL, pi.m_fd, nullptr) != -1;	200,065✔
177	#endif
178	}
179
180	auto io_notifier::unwatch_timer(const timer_handle& timer) -> bool	170✔
181	{
182	#if defined(__FreeBSD__) \|\| defined(__APPLE__) \|\| defined(__OpenBSD__) \|\| defined(__NetBSD__)
183	auto event_data = event_t{};
184	EV_SET(&event_data, timer.m_fd, EVFILT_TIMER, EV_DELETE, 0, 0, nullptr);
185	return ::kevent(m_fd, &event_data, 1, nullptr, 0, nullptr) != -1;
186	#elif defined(__linux__)
187	// Setting these values to zero disables the timer.
188	itimerspec ts{};	170✔
189	ts.it_value.tv_sec = 0;	170✔
190	ts.it_value.tv_nsec = 0;	170✔
191	return ::timerfd_settime(timer.m_fd, 0, &ts, nullptr) != -1;	170✔
192	#endif
193	}
194
195	auto io_notifier::next_events(std::chrono::milliseconds timeout)	56,500,584✔
196	-> std::vector<std::pair<detail::poll_info*, coro::poll_status>>
197	{
198	auto ready_events = std::vector<std::pair<detail::poll_info*, coro::poll_status>>();	56,500,584✔
199	auto ready_set = std::array<event_t, m_max_events>{};	56,499,011✔
200	#if defined(__FreeBSD__) \|\| defined(__APPLE__) \|\| defined(__OpenBSD__) \|\| defined(__NetBSD__)
201	const auto timeout_as_secs = std::chrono::duration_cast<std::chrono::seconds>(timeout);
202	auto timeout_spec = ::timespec{
203	.tv_sec = timeout_as_secs.count(),
204	.tv_nsec = std::chrono::duration_cast<std::chrono::nanoseconds>(timeout - timeout_as_secs).count(),
205	};
206	const std::size_t num_ready = ::kevent(m_fd, nullptr, 0, ready_set.data(), ready_set.size(), &timeout_spec);
207	for (std::size_t i = 0; i < num_ready; i++)
208	{
209	ready_events.emplace_back(
210	static_cast<detail::poll_info*>(ready_set[i].udata), io_notifier::event_to_poll_status(ready_set[i]));
211	}
212	#elif defined(__linux__)
213	int num_ready = ::epoll_wait(m_fd, ready_set.data(), ready_set.size(), timeout.count());	56,499,011✔
214	for (int i = 0; i < num_ready; ++i)	113,163,724✔
215	{
216	ready_events.emplace_back(	56,670,995✔
217	static_cast<detail::poll_info*>(ready_set[i].data.ptr), io_notifier::event_to_poll_status(ready_set[i]));	56,682,965✔
218	}
219	#endif
220	return ready_events;	112,961,518✔
NEW 221	}	×
222
223	auto io_notifier::event_to_poll_status(const event_t& event) -> poll_status	56,668,981✔
224	{
225	#if defined(__FreeBSD__) \|\| defined(__APPLE__) \|\| defined(__OpenBSD__) \|\| defined(__NetBSD__)
226	if (event.filter & EVFILT_READ \|\| event.filter & EVFILT_WRITE)
227	{
228	return poll_status::event;
229	}
230	else if (event.flags & EV_ERROR)
231	{
232	return poll_status::error;
233	}
234	else if (event.flags & EV_EOF)
235	{
236	return poll_status::closed;
237	}
238	throw std::runtime_error{"invalid kqueue state"};
239	#elif defined(__linux__)
240	if (event.events & static_cast<uint32_t>(poll_op::read) \|\| event.events & static_cast<uint32_t>(poll_op::write))	56,668,981✔
241	{
242	return poll_status::event;	56,668,981✔
243	}
NEW 244	else if (event.events & EPOLLERR)	×
245	{
NEW 246	return poll_status::error;	×
247	}
NEW 248	else if (event.events & EPOLLRDHUP \|\| event.events & EPOLLHUP)	×
249	{
NEW 250	return poll_status::closed;	×
251	}
NEW 252	throw std::runtime_error{"invalid epoll state"};	×
253	#endif
254	}
255
256	} // namespace coro

jbaldwin / libcoro / 14957747050

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