Asio. Error: "The I/o operation has been aborted because of either a thread exit or an application requests"
I am trying to create an async server using asio but when the acceptor calls async_accept function, I get this error The I/o operation has been aborted because of either a thread exit or an application requests, which doesn't let the process continue. I have tried to change the port but it doesn't work. When I try to make a sync server so it works. Also I use asio as a separate lib from boost.
Here's the function where I get the error:
void startAsyncAccept()
{
acceptor.async_accept(socket, [&](const asio::error_code& error)
{
if(!error)
{
std::cout << "Client is connected" << "\n";
startAsyncRead();
}
else
{
std::cerr << "It is: " << error.message() << '\n';
return 0;
}
startAsyncAccept();
}
);
}
Ans here's my async server:
class Server
{
public:
Server(asio::io_context& io_context) : acceptor(io_context, asio::ip::tcp::endpoint(asio::ip::address::from_string("127.0.0.1"), 2000)), socket(io_context)
{
startAsyncAccept();
}
~Server() {}
private:
// Members
asio::ip::tcp::acceptor acceptor;
asio::ip::tcp::socket socket;
asio::streambuf buffer;
std::vector<asio::ip::tcp::socket> sockets;
// Functions
void startAsyncAccept()
{
acceptor.async_accept(socket, [&](const asio::error_code& error)
{
if(!error)
{
std::cout << "Client is connected" << "\n";
startAsyncRead();
}
else
{
std::cerr << "It is: " << error.message() << '\n';
return 0;
}
startAsyncAccept();
}
);
}
void startAsyncRead()
{
asio::async_read_until(socket, buffer, '\n', [&](const asio::error_code& error, size_t length)
{
if(!error)
{
std::string message(asio::buffers_begin(buffer.data()), asio::buffers_begin(buffer.data()) + length);
std::cout << "Received from client: " << message;
for(auto& clients : sockets)
{
asio::async_write(clients, asio::buffer(message), [&](const asio::error_code& error, size_t length)
{
if(error)
{
std::cerr << "Failed to write to client: " << error.message() << "\n";
}
}
);
}
buffer.consume(length);
startAsyncRead();
}
else
{
std::cout << "Client disconnected." << std::endl;
removeClient();
}
}
);
}
void startAsyncWrite()
{
asio::async_write(socket, asio::buffer("Connected to chat Server. \n"), [&](const asio::error_code& error, size_t length)
{
if(error)
{
std::cerr << "Failed to write to the client: " << error.message() << "\n";
removeClient();
}
}
);
}
void removeClient()
{
auto it = std::find_if(sockets.begin(), sockets.end(),
[&](const auto& client_socket)
{
return &client_socket == &socket;
});
if (it != sockets.end())
{
sockets.erase(it);
}
}
};
First of all, there's Undefined Behaviour because the lambda misses a return on some paths. This seems to be a copy-paste error of some sort, so I'll just remove the return 0; that was meaningless.
The Real Problem
The real problem is that you use the same socket on each accept. That means that you reset the socket that you started the first asyncRead on as soon as you call startAsyncAccept again. That causes the asio::error::operation_aborted error code.
Simply don't use the same socket. Move it into a "session" object e.g. Or, as you seem to have wanted to do, put them in your collection of sockets.
Fixing?
Ironically, it is hard to know how you wanted to use sockets since it is literally unused in your current code (except for the loops that will never iterate, because sockets is always empty).
What's worse, it's only good that didn't work, because otherwise you get more UB, because std::vector will reallocate to grow capacity, meaning that any references to existing sockets are invalidated.
The usual way to deal with this is not to have a collection of sockets, but rather collection of (weak) shared pointers to sessions objects. For simplicity, let me fix the reference stability here by changing std::vector to std::list and removing the problematic socket variable.
tcp::socket socket;
asio::streambuf buffer;
std::vector<tcp::socket> sockets;
Becomes
asio::streambuf buffer;
std::list<tcp::socket> sockets;
There's big problems as well with async_write as
- you use a local variable
messageas the buffer. This is UB again, because its lifetime ends before the write operations complete - you MUST avoid overlapping calls. Because you don't control when
async_read_untilcompletes (and you might have an arbitrary number of connected clients), you cannot be sure no writes overlap. The only solution there is to have an "outbox" queue - typically per client, for independent operation of each connection.
As a sidenote, it's unclear what startAsyncWrite was supposed to do (it was unused, and doesn't help to solve the problem just described). Same with removeClient which would never do anything even if you had filled sockets, it would just cause UB due to the invalidation of socket references. I dropped those two for now.
Finally, you use a streambuf, where it seems you might just use a dynamic string buffer directly. In any case, put the consume closer to the "consume" so that you don't risk missing it under some error condition.
Simplified V1
This simplified version removes the problems that you ask about as well as the lifetime of the message:
class Server {
public:
Server(asio::io_context& io_context) //
: acceptor(io_context, {{}, 2000}) {
startAsyncAccept();
}
private:
// Members
tcp::acceptor acceptor;
asio::streambuf buffer;
std::list<tcp::socket> sockets;
void startAsyncAccept() {
acceptor.async_accept([&](error_code error, tcp::socket accepted) {
if (!error) {
auto& client = sockets.emplace_back(std::move(accepted));
std::cout << "Client is connected\n";
startAsyncRead(client);
} else {
std::cerr << "It is: " << error.message() << "\n";
}
startAsyncAccept();
});
}
void startAsyncRead(tcp::socket& client) {
asio::async_read_until(client, buffer, '\n', [&](error_code error, size_t n) {
if (!error) {
auto f = asio::buffers_begin(buffer.data()), l = f + static_cast<ptrdiff_t>(n);
auto message = std::make_shared<std::string>(f, l);
buffer.consume(n);
std::cout << "Received from client: " << message;
for (auto& peer : sockets) {
// SEHE: TODO FIXME serializing writes per peer!
asio::async_write(
peer, asio::buffer(*message), [&, message](error_code error, size_t /*length*/) {
if (error) {
std::cerr << "Failed to write to client: " << error.message() << "\n";
}
});
}
startAsyncRead(client);
} else {
std::cout << "Client disconnected." << std::endl;
// SEHE: TODO ERASE
}
});
}
};
Fixed: Client Sessions
Some more work to fix the other problems:
[after breakfast]
UPDATE You absolutely need a separate Session, because I missed earlier (before breakfast) that you were also falsely using the same streambuf instance for all connections... A Session type makes for a logical place to group the incoming buffer, outgoing buffer queue and socket.
We use enable_shared_from_this together with shared_from_this() to get automatic lifetime management when a connection gets bad/closed.
Note that it all comes down to correct Separation Of Concerns: The Server manages the listening, the ChatRoom manages connected clients, the Session manages a single client.
#include <boost/asio.hpp>
#include <deque>
#include <iostream>
#include <list>
#ifndef STANDALONE_ASIO
namespace asio = boost::asio;
using boost::system::error_code;
#else
using asio::error_code;
#endif
using asio::ip::tcp;
struct Session;
using SessionPtr = std::shared_ptr<Session>;
using Handle = std::weak_ptr<Session>;
struct ChatRoom {
std::list<Handle> clients_;
void garbage_collect();
void broadcast(std::string_view message);
// just for example:
void direct_message(std::string recipient, std::string_view message);
};
struct Session : std::enable_shared_from_this<Session> {
Session(tcp::socket s, ChatRoom& room) : socket_(std::move(s)), room_(room) {}
void start() { readLoop(); }
void send(std::string_view message){
outbox_.emplace_back(message);
if (outbox_.size() == 1)
writeLoop();
}
private:
tcp::socket socket_;
ChatRoom& room_;
asio::streambuf incoming_;
std::deque<std::string> outbox_;
void readLoop() {
asio::async_read_until(
socket_, incoming_, '\n', [&, self = shared_from_this()](error_code ec, size_t n) {
if (ec) {
std::cout << "Client disconnect (" << ec.message() << ")" << std::endl;
return;
}
auto f = asio::buffers_begin(incoming_.data()), l = f + static_cast<ptrdiff_t>(n);
room_.broadcast(std::string(f, l));
incoming_.consume(n);
readLoop();
});
}
void writeLoop() {
if (outbox_.empty())
return;
asio::async_write( //
socket_, asio::buffer(outbox_.front()),
[this, self = shared_from_this()](error_code ec, size_t /*length*/) {
outbox_.pop_front();
if (ec)
std::cerr << "Failed to write to client: " << ec.message() << "\n";
else
writeLoop();
});
}
};
class Server {
public:
Server(asio::any_io_executor ex) : acceptor_(ex, {{}, 2000}) { acceptLoop(); }
private:
tcp::acceptor acceptor_;
ChatRoom room_;
void acceptLoop() {
room_.garbage_collect(); // optionally prevent dead connections piling up
acceptor_.async_accept([&](error_code ec, tcp::socket accepted) {
if (!ec) {
auto sess = std::make_shared<Session>(std::move(accepted), room_);
room_.clients_.push_back(sess);
std::cout << "Client is connected\n";
sess->start();
} else {
std::cerr << "Accept error: " << ec.message() << "\n";
}
acceptLoop();
});
}
};
void ChatRoom::garbage_collect() {
clients_.remove_if(std::mem_fn(&Handle::expired));
}
void ChatRoom::broadcast(std::string_view message) {
for (auto& handle : clients_) {
if (auto peer = handle.lock()) {
peer->send(message);
}
}
}
using namespace std::chrono_literals;
int main() {
asio::io_context io;
Server chat(io.get_executor());
io.run_for(30s);
}
Testing with a number of clients:
for a in {1..10}; do (
sleep 1.$RANDOM
echo "hello from $a"
sleep 1.$RANDOM
echo "bye from $a"
) | nc 127.0.0.1 2000 -w3 | (while read line; do echo "Client $a received '$line'"; done) & done
Prints output like:
Also live on Coliru :)
- Casting Pointer array from one type to another
- Minimizing the number of malloc() calls improves performance?
- Difference between curl/curl.h, libcurl, libcurl4-openssl-dev, libcurl4-nss-dev libraries?
- Why doesn't my manually entered double value match the expected value in C?
- From 2D Arrays to 2x2 Matrix in C
- How to determine the value of socket listen() backlog parameter?
- What does #define do in the Compiler?
- Deleting selected element in a doubly linked list
- Writing to qemu RISCV UART using c
- Bubble sort does not sort as expected
- How to store and then print a 2d character/string array horizontally for 5 times?
- Fast CRC algorithm?
- Assertion —to a function pointer— failed
- Learning c, my code runs but doesn't displayed print?
- Strange size change in core dump file
- Rust's .o file format not recognized by GCC
- How to declare a constant "mutable pointer" to an immutable block of memory
- Printing all environment variables in C / C++
- What is the difference between getch() and getchar()?
- Filemanagement in C: Unsuccessful attempt in creating/overwriting a file using fopen
- Get pixel color at mouse position fastest way
- Valid VISA: 4222222222222 outputted as invalid
- Unique Words Counter in C
- Confused about casting and order of operations
- What is the closest thing Windows has to fork()?
- Why No Variable Declarations in Boolean Expressions in C?
- Search Algorithm to find the k lowest values in a list (selection algorithm/problem)
- What is the need of hh and h format specifiers?
- can't load bmp file with SDL_LoadBMP in CodeBlocks
- Need help to solve function issue