I am trying to build a simple IPC protocol using Boost Asio where the server side will be sending a struct that contains a vector<uint8_t> to the client. I was suggested to use a scatter/gather IO approach, but I can't get it working, as it seems the client is only receiving part of the data it is expecting and it keeps waiting indefinitely for the rest of the data to arrive even though it should already be there.
This is what I have right now:
// File: client.cpp
#include <iostream>
#include <vector>
#include <boost/asio.hpp>
#include "ipc_common.hpp"
namespace ba = boost::asio;
using boost::asio::ip::tcp;
int main(int argc, char *argv[])
{
ba::io_context io;
std::vector<std::string> args(argv, argv + argc);
switch (args.size()) {
case 1:
args = {args.at(0), "localhost", "6869"};
break;
case 2:
args = {args.at(0), args.at(1), "6869"};
break;
case 3:
args = {args.at(0), args.at(1), args.at(2)};
break;
default:
std::clog << "usage: " << args.at(0) << " [host = localhost] [port = 6869]" << std::endl;
return 1;
}
try {
propertiesPacket properties;
properties.val1 = 9;
properties.val2 = 45;
tcp::socket socket(io);
tcp::resolver resolver(io);
connect(socket, resolver.resolve(args.at(1), args.at(2)));
write(socket, ba::buffer(&properties, sizeof(properties)));
uint16_t responseSize {};
ba::read(socket, ba::buffer(&responseSize, sizeof(uint16_t)));
std::clog << "client responseSize: " << responseSize << std::endl;
processedData response {};
std::vector<ba::mutable_buffer> responseBuffers {
ba::buffer(&response.size, sizeof(uint16_t)),
ba::buffer(&response.values, responseSize - sizeof(uint8_t))
};
ba::read(socket, responseBuffers);
std::clog << response.serialize();
return 0;
} catch (std::exception &e) {
std::clog << e.what() << std::endl;
return 1;
}
}
// File: server.cpp
#include <vector>
#include <boost/asio.hpp>
#include "ipc_common.hpp"
namespace ba = boost::asio;
using boost::asio::ip::tcp;
using boost::system::error_code;
using TCPSocket = tcp::socket;
class ServerConnection
: public std::enable_shared_from_this<ServerConnection>
{
public:
ServerConnection(TCPSocket socket)
: socket_(std::move(socket))
{ }
void start()
{
std::clog << __PRETTY_FUNCTION__ << std::endl;
doRead();
}
private:
void doRead()
{
std::clog << __PRETTY_FUNCTION__ << std::endl;
auto self(shared_from_this());
socket_.async_read_some(ba::buffer(&properties_, sizeof(properties_)),
[this, self](error_code ec, std::size_t length)
{
std::clog << "received " << length << std::endl;
if (!ec) {
processData();
std::vector<ba::const_buffer> msg {
ba::buffer(&filePacketSize_, sizeof(uint16_t)),
ba::buffer(&filePacket_.val, sizeof(filePacket_.val)),
ba::buffer(&filePacket_.values, sizeof(filePacket_.values))};
std::clog << "filePacketSize_: " << filePacketSize_ << std::endl;
ba::async_write(socket_, msg,
[this, self = shared_from_this()](error_code ec, std::size_t length)
{
std::clog << "written " << length << std::endl;
if (!ec) doRead();
});
}
});
}
void processData()
{
filePacket_.val = properties_.val1;
// Just for demonstration, we fill the vector with random values
std::random_device rd;
std::mt19937 re(rd()) ;
std::uniform_int_distribution<uint8_t> dist(0, 255);
for (size_t i {}; i < filePacket_.val; ++i) {
processedData.values.push_back(dist(re));
}
}
TCPSocket socket_;
propertiesPacket properties_;
processedData filePacket_;
uint16_t filePacketSize_;
};
class Server
{
public:
using IOContext = ba::io_context;
using TCPAcceptor = tcp::acceptor;
Server(IOContext& io, uint16_t port)
: socket_(io),
acceptor_(io, {tcp::v4(), port})
{
doAccept();
}
private:
void doAccept()
{
std::clog << __PRETTY_FUNCTION__ << std::endl;
acceptor_.async_accept(socket_,
[this](error_code ec)
{
if (!ec) {
std::clog << "Accepted " << socket_.remote_endpoint() << std::endl;
std::make_shared<ServerConnection>(std::move(socket_))->start();
doAccept();
}
else {
std::clog << "Accept " << ec.message() << std::endl;
}
});
}
TCPSocket socket_;
TCPAcceptor acceptor_;
};
int main(int argc, char* argv[])
{
std::vector<std::string> args(argv, argv + argc);
switch (args.size()) {
case 1:
args = {args.at(0), "6869"};
break;
case 2:
args = {args.at(0), args.at(1)};
break;
default:
std::clog << "usage: " << args.at(0) << " [port = 6869]" << std::endl;
return 1;
}
try {
ba::io_context io;
Server server(io, std::stoi(args.at(1)));
io.run();
} catch (std::exception &e) {
std::clog << e.what() << std::endl;
return 1;
}
return 0;
}
// File: ipc_common.hpp
#include <cstdint>
#include <vector>
#include <sstream>
#include <string>
struct propertiesPacket
{
uint8_t val1;
uint8_t val2;
};
struct processedData
{
uint8_t val;
std::vector<uint8_t> values;
std::string serialize()
{
std::stringstream sstream;
sstream << "val: " << (unsigned int)val << std::endl;
for (const auto &i : values)
{
sstream << i << " ";
}
sstream << std::endl;
return sstream.str();
}
};
What am I doing wrong?
