9921718395

Committed 05 Jul 2024 06:29PM UTC coverage: 84.498% (+0.9%) from 83.609%

Build # 9921718395

Build Type

push

github

Committed by

web-flow

Commit Message

coro::thread_pool high cpu usage when tasks < threads (#265)

* coro::thread_pool high cpu usage when tasks < threads

The check for m_size > 0 was keeping threads awake in a spin state until
all tasks completed. This correctl now uses m_queue.size() behind the
lock to correctly only wake up threads on the condition variable when
tasks are waiting to be processed.

* Fix deadlock with task_container and tls::client with the client's
  destructor scheduling a tls cleanup task, the task_container's lock
was being locked twice when the cleanup task was being destroyed.

Closes #262

* Adjust when task_container's user_task is deleted

It is now deleted inline in make_user_task so any destructors that get
invoked that possibly schedule more coroutines do not cause a deadlock

* io_scheduler is now std::enable_shared_from_this

Run Details

51 of 64 new or added lines in 5 files covered. (79.69%)

51 existing lines in 4 files now uncovered.

1379 of 1632 relevant lines covered (84.5%)

4234968.15 hits per line

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

52.24

/src/net/socket.cpp

#include "coro/net/socket.hpp"

namespace coro::net
{
auto socket::type_to_os(type_t type) -> int
{
    switch (type)
    {
        case type_t::udp:
            return SOCK_DGRAM;
        case type_t::tcp:
            return SOCK_STREAM;
        default:
            throw std::runtime_error{"Unknown socket::type_t."};
    }
}

auto socket::operator=(const socket& other) noexcept -> socket&
{
    this->close();
    this->m_fd = dup(other.m_fd);
    return *this;
}

auto socket::operator=(socket&& other) noexcept -> socket&
{
    if (std::addressof(other) != this)
    {
        m_fd = std::exchange(other.m_fd, -1);
    }

    return *this;
}

auto socket::blocking(blocking_t block) -> bool
{
    if (m_fd < 0)
    {
        return false;
    }

    int flags = fcntl(m_fd, F_GETFL, 0);
    if (flags == -1)
    {
        return false;
    }

    // Add or subtract non-blocking flag.
    flags = (block == blocking_t::yes) ? flags & ~O_NONBLOCK : (flags | O_NONBLOCK);

    return (fcntl(m_fd, F_SETFL, flags) == 0);
}

auto socket::shutdown(poll_op how) -> bool
{
    if (m_fd != -1)
    {
        int h{0};
        switch (how)
        {
            case poll_op::read:
                h = SHUT_RD;
                break;
            case poll_op::write:
                h = SHUT_WR;
                break;
            case poll_op::read_write:
                h = SHUT_RDWR;
                break;
        }

        return (::shutdown(m_fd, h) == 0);
    }
    return false;
}

auto socket::close() -> void
{
    if (m_fd != -1)
    {
        ::close(m_fd);
        m_fd = -1;
    }
}

auto make_socket(const socket::options& opts) -> socket
{
    socket s{::socket(static_cast<int>(opts.domain), socket::type_to_os(opts.type), 0)};
    if (s.native_handle() < 0)
    {
        throw std::runtime_error{"Failed to create socket."};
    }

    if (opts.blocking == socket::blocking_t::no)
    {
        if (s.blocking(socket::blocking_t::no) == false)
        {
            throw std::runtime_error{"Failed to set socket to non-blocking mode."};
        }
    }

    return s;
}

auto make_accept_socket(const socket::options& opts, const net::ip_address& address, uint16_t port, int32_t backlog)
    -> socket
{
    socket s = make_socket(opts);

    int sock_opt{1};
    if (setsockopt(s.native_handle(), SOL_SOCKET, SO_REUSEADDR | SO_REUSEPORT, &sock_opt, sizeof(sock_opt)) < 0)
    {
        throw std::runtime_error{"Failed to setsockopt(SO_REUSEADDR | SO_REUSEPORT)"};
    }

    sockaddr_in server{};
    server.sin_family = static_cast<int>(opts.domain);
    server.sin_port   = htons(port);
    server.sin_addr   = *reinterpret_cast<const in_addr*>(address.data().data());

    if (bind(s.native_handle(), (struct sockaddr*)&server, sizeof(server)) < 0)
    {
        throw std::runtime_error{"Failed to bind."};
    }

    if (opts.type == socket::type_t::tcp)
    {
        if (listen(s.native_handle(), backlog) < 0)
        {
            throw std::runtime_error{"Failed to listen."};
        }
    }

    return s;
}

} // namespace coro::net

1	#include "coro/net/socket.hpp"
2
3	namespace coro::net
4	{
5	auto socket::type_to_os(type_t type) -> int	311✔
6	{
7	switch (type)	311✔
8	{
9	case type_t::udp:	4✔
10	return SOCK_DGRAM;	4✔
11	case type_t::tcp:	308✔
12	return SOCK_STREAM;	308✔
13	default:
14	throw std::runtime_error{"Unknown socket::type_t."};
15	}
16	}
17
18	auto socket::operator=(const socket& other) noexcept -> socket&	×
19	{
20	this->close();	×
UNCOV 21	this->m_fd = dup(other.m_fd);	×
22	return *this;	×
23	}
24
25	auto socket::operator=(socket&& other) noexcept -> socket&	×
26	{
UNCOV 27	if (std::addressof(other) != this)	×
28	{
UNCOV 29	m_fd = std::exchange(other.m_fd, -1);	×
30	}
31
32	return *this;	×
33	}
34
35	auto socket::blocking(blocking_t block) -> bool	594✔
36	{
37	if (m_fd < 0)	594✔
38	{
UNCOV 39	return false;	×
40	}
41
42	int flags = fcntl(m_fd, F_GETFL, 0);	594✔
43	if (flags == -1)	619✔
44	{
UNCOV 45	return false;	×
46	}
47
48	// Add or subtract non-blocking flag.
49	flags = (block == blocking_t::yes) ? flags & ~O_NONBLOCK : (flags \| O_NONBLOCK);	619✔
50
51	return (fcntl(m_fd, F_SETFL, flags) == 0);	619✔
52	}
53
54	auto socket::shutdown(poll_op how) -> bool	×
55	{
56	if (m_fd != -1)	×
57	{
58	int h{0};	×
59	switch (how)	×
60	{
61	case poll_op::read:	×
62	h = SHUT_RD;	×
UNCOV 63	break;	×
UNCOV 64	case poll_op::write:	×
65	h = SHUT_WR;	×
UNCOV 66	break;	×
67	case poll_op::read_write:	×
UNCOV 68	h = SHUT_RDWR;	×
UNCOV 69	break;	×
70	}
71
UNCOV 72	return (::shutdown(m_fd, h) == 0);	×
73	}
UNCOV 74	return false;	×
75	}
76
77	auto socket::close() -> void	2,540✔
78	{
79	if (m_fd != -1)	2,540✔
80	{
81	::close(m_fd);	626✔
82	m_fd = -1;	626✔
83	}
84	}	2,540✔
85
86	auto make_socket(const socket::options& opts) -> socket	316✔
87	{
88	socket s{::socket(static_cast<int>(opts.domain), socket::type_to_os(opts.type), 0)};	316✔
89	if (s.native_handle() < 0)	297✔
90	{
91	throw std::runtime_error{"Failed to create socket."};	×
92	}
93
94	if (opts.blocking == socket::blocking_t::no)	294✔
95	{
96	if (s.blocking(socket::blocking_t::no) == false)	291✔
97	{
UNCOV 98	throw std::runtime_error{"Failed to set socket to non-blocking mode."};	×
99	}
100	}
101
102	return s;	316✔
UNCOV 103	}	×
104
105	auto make_accept_socket(const socket::options& opts, const net::ip_address& address, uint16_t port, int32_t backlog)	18✔
106	-> socket
107	{
108	socket s = make_socket(opts);	18✔
109
110	int sock_opt{1};	18✔
111	if (setsockopt(s.native_handle(), SOL_SOCKET, SO_REUSEADDR \| SO_REUSEPORT, &sock_opt, sizeof(sock_opt)) < 0)	18✔
112	{
UNCOV 113	throw std::runtime_error{"Failed to setsockopt(SO_REUSEADDR \| SO_REUSEPORT)"};	×
114	}
115
116	sockaddr_in server{};	18✔
117	server.sin_family = static_cast<int>(opts.domain);	18✔
118	server.sin_port = htons(port);	18✔
119	server.sin_addr = reinterpret_cast<const in_addr>(address.data().data());	18✔
120
121	if (bind(s.native_handle(), (struct sockaddr*)&server, sizeof(server)) < 0)	18✔
122	{
123	throw std::runtime_error{"Failed to bind."};	×
124	}
125
126	if (opts.type == socket::type_t::tcp)	18✔
127	{
128	if (listen(s.native_handle(), backlog) < 0)	15✔
129	{
UNCOV 130	throw std::runtime_error{"Failed to listen."};	×
131	}
132	}
133
134	return s;	36✔
UNCOV 135	}	×
136
137	} // namespace coro::net

jbaldwin / libcoro / 9921718395

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