ripose-jp
commented
10 months ago Here's a "minimal" example I could create.
#include <coro/coro.hpp>
#include <iostream>
int main()
{
auto scheduler = std::make_shared<coro::io_scheduler>(coro::io_scheduler::options{
.execution_strategy = coro::io_scheduler::execution_strategy_t::process_tasks_inline
});
constexpr size_t REQUEST_SIZE{100 * 1024 * 1024};
auto make_bogus_read_task = [](std::shared_ptr<coro::net::tcp_client> client) -> coro::task<void>
{
using namespace std::chrono_literals;
co_await client->poll(coro::poll_op::read, 10s);
};
auto make_server_task = [&]() -> coro::task<std::string>
{
coro::net::tcp_server server{scheduler};
co_await scheduler->schedule();
auto poll_status = co_await server.poll();
if (poll_status != coro::poll_status::event)
{
co_return {};
}
std::shared_ptr<coro::net::tcp_client> client =
std::make_shared<coro::net::tcp_client>(server.accept());
if (!client->socket().is_valid())
{
co_return {};
}
scheduler->schedule(make_bogus_read_task(client));
std::string request(REQUEST_SIZE, 0xAA);
std::span<const char> remaining{request};
do
{
auto [send_status, r] = client->send(remaining);
if (send_status != coro::net::send_status::ok)
{
co_return {};
}
if (r.empty())
{
break;
}
remaining = r;
auto pstatus = co_await client->poll(coro::poll_op::write);
if (pstatus != coro::poll_status::event)
{
co_return {};
}
} while (true);
co_return request;
};
auto make_client_task = [&]() -> coro::task<std::string>
{
co_await scheduler->schedule();
coro::net::tcp_client client{scheduler};
auto status = co_await client.connect();
if (status != coro::net::connect_status::connected)
{
co_return {};
}
std::string response(REQUEST_SIZE, '\0');
std::span<char> remaining{response};
do
{
auto poll_status = co_await client.poll(coro::poll_op::read);
if (poll_status != coro::poll_status::event)
{
co_return {};
}
auto [recv_status, r] = client.recv(remaining);
if (recv_status != coro::net::recv_status::ok)
{
co_return {};
}
remaining = remaining.subspan(r.size_bytes());
} while (!remaining.empty());
co_return response;
};
auto result = coro::sync_wait(
coro::when_all(make_server_task(), make_client_task())
);
std::string request = std::move(std::get<0>(result).return_value());
std::string response = std::move(std::get<1>(result).return_value());
if (request.empty())
{
std::cout << "Request was empty\n";
}
else if (response.empty())
{
std::cout << "Response was empty\n";
}
else if (request != response)
{
std::cout << "Request != Response\n";
}
else
{
std::cout << "Request == Response\n";
}
}
jbaldwin
I'm writing a system where a
coro::net::tcp_serverlistens for incoming connections and does two things when receiving an incoming connection:poll(coro::poll_op::read),recv(), and puts structured data in aring_buffer. This coroutine lives in its owntask_container.poll(coro::poll_op::write)andsend()when outgoing data is ready.The issue arises at the
pollcall. I believe thatcoro::net::tcp_clientboth share an epoll file descriptor, so when two coroutines concurrently try to poll for both EPOLLIN and EPOLLOUT separately, it is possible forepoll_ctlto fail withEEXISTright here.I don't think that epolling for data concurrently should cause error conditions, but I'm having trouble thinking of ways to fix the current API design.
My first though was separating
poll_status::eventintopoll_status::readandpoll_status::writeso that two coroutines could concurrently poll. If a reader coroutine gets apoll_status::writeit can ignore it and vice versa. The problem with this approach is aftertcp_client::poll()returns, I believe it unregisters that file descriptor fromepollwhich will leave the other coroutine in an undefined state. Some sort of ref count would need to be included inpoll_infoor something to prevent this. It is also a breaking API change.Currently I've modified this code which seems to fix the issue in the case of an
coro::io_schedulerusing inline task processing. I have tested it using a threadpool yet. Regardless, I have misgivings about it's correctness, so I'm interested in finding a better solution to the problem.