CODEWITHSUNDEEP
c++network-programmingboostipcboost-asio
Spiwocoal
Spiwocoal

Reputation: 37

How to use Scatter/Gather IO in Boost Asio for sending a struct containing a vector?

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?

Upvotes: 1

Views: 960

Answers (1)

sehe
sehe

Reputation: 393134

The sample seems corrupted.

  1. For one, args.at(3) and args.at(4) will by definition always throw, because by definition the switch statement earlier will always exit the client when there are more than 2 command line arguments (default:).

  2. Secondly, the client read uses &response.size but no such member exists at all.

  3. Thirdly, server processData uses a .val property of procesedData which isn't even a member (it's a type, likely should be filePacket_.val instead).

  4. Fourthly, it assigns that from properties_.val; which ALSO doesn't exist at all (there's only val1 and val2).

  5. Next up, rd isn't used to initialize the URBG (random engine, re). Instead it calls an unknown identifier named random_device(). Likely ought to be rd() instead.

  6. Again, where it sais processedData.val you probably meant filePacket_.val

  7. And where you write processedData.push_back(...) you probably meant to say filePacket_.values.push_back(...)...

  8. There's a spurious ; behind void doAccept() in the Server

  9. By contrast, the ; is missing after each struct definition in ipc_common.hpp

  10. The processedData struct defines a serialize() method that is never used. It also uses a C-style cast where static_cast<unsigned>(val) would be safe.

  11. Weirdly, the server "parses" args, and provides an optional default value BUT it never uses that. Instead, it uses argv[1] without checking argc at all. Oops.

That all aside, now comes the confusing part: how did you want the values to be written? This is not correct:

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))};

values is a std::vector<> so you cannot hope to use it in a bitwise way. It'll just invoke Undefined Behaviour.

Besides, it's pretty unclear why filePacketSize_ is being written (it's never even assigned, or even initialized to a determinate value).

On the client side you read a responseSize as if one would be sent... Maybe you want to keep those two in sync.

Suggested Appraoch

I'd do away with the separate size value(s), since a vector already keeps track of that. I'd also make sure your processData doesn't always push_back because the vector would always keep growing.

I'd make a protocol that actually sends the message size before the message itself, and makes sure it's correct.

Let's also make the random data naturally printable (a..z) for simplicity:

void processData()
{
    // Just for demonstration, we fill the vector with random characters
    std::mt19937 re(std::random_device{}());
    std::uniform_int_distribution<uint8_t> dist('a', 'z');
    filePacket_.values.clear();
    std::generate_n(back_inserter(filePacket_.values), properties_.val1,
                    [&] { return dist(re); });
}

Then in writing, let's do:

processData();

size_t length[] { filePacket_.values.size() };

std::vector<ba::const_buffer> msg{
    ba::buffer(length),
    ba::buffer(filePacket_.values)};

Note how, again, we avoid manually specifying any buffer sizes. Also, we let the library figure out that values is a vector of POD elements and do the math to convert the calculate the correct start address and buffer size for the element data.

On the client side, we do the inverse:

size_t length = 0;
ba::read(socket, ba::buffer(&length, sizeof(length)));

response.values.resize(length);
ba::read(socket, ba::buffer(response.values));

(Here we can't avoid writing sizeof(length) without getting more clumsy than I'd like).

Full Demo

  • File ipc_common.hpp

     // File: ipc_common.hpp
 #include <cstdint>
 #include <sstream>
 #include <string>
 #include <vector>

 struct propertiesPacket {
     uint8_t val1;
     uint8_t val2;
 };

 struct processedData {
     std::vector<uint8_t> values;
 };

  • File server.cpp

     #include <boost/asio.hpp>
 #include <vector>
 #include <iostream>
 #include <random>

 #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();

                     size_t length[] { filePacket_.values.size() };

                     std::vector<ba::const_buffer> msg{
                         ba::buffer(length), ba::buffer(filePacket_.values)};

                     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()
         {
             // Just for demonstration, we fill the vector with random characters
             std::mt19937 re(std::random_device{}());
             std::uniform_int_distribution<uint8_t> dist('a', 'z');
             filePacket_.values.clear();
             std::generate_n(back_inserter(filePacket_.values), properties_.val1,
                             [&] { return dist(re); });
         }

     TCPSocket socket_;

     propertiesPacket properties_;
     processedData    filePacket_;
 };

 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.push_back("6869"); break;
         case 2: 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 const &e) {
         std::clog << e.what() << std::endl;
         return 1;
     }
 }

  • 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.push_back("localhost"); [[fallthrough]];
         case 2: args.push_back("6869");      [[fallthrough]];
         case 3: args.push_back("42");        [[fallthrough]];
         case 4: args.push_back("99");        [[fallthrough]];
         case 5: break;
         default:
             std::clog << "usage: " << args.at(0)
                       << " [host = localhost] [port = 6869] [val1=42] [val2=99]"
                       << std::endl;
             return 1;
     }

     try {
         propertiesPacket properties;
         properties.val1 = std::stoul(args.at(3));
         properties.val2 = std::stoul(args.at(4));

         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)));

         processedData response{};
         {
             size_t length = 0;
             ba::read(socket, ba::buffer(&length, sizeof(length)));

             response.values.resize(length);
         }

         std::clog << "client response size: " << response.values.size() << std::endl;
         ba::read(socket, ba::buffer(response.values));

         std::clog.write(reinterpret_cast<char const*>(response.values.data()),
                         response.values.size()) << "\n";
         // return 0;
     } catch (std::exception &e) {
         std::clog << e.what() << std::endl;
         return 1;
     }
 }

Demo output:

enter image description here

Portability

You should probably keep byte ordering in mind as well. You could consider using JSON or another Well Known serialization format.

Upvotes: 2

Related Questions