boost::async read write stuck after one read and two writes

Question

I trying to build a server-client application the server-side is c++ with boost::asio async but when I'm running the application it gets stuck after 1 reading and two writes. this is the relevant code when I'm removing the do_write function call I'm getting all 5 messages that I'm sending

EDIT when I'm running with a debugger I'm able to receive all five messages and send all the response

void start()
  {
    std::thread t1([this](){do_write();});
    t1.detach();
    std::thread t2([this](){do_read();});
    t2.detach();
  }

void do_write()
  {
    auto self(shared_from_this());
    Message msg;
    while(order_response_queue_.empty())
    {}
    auto order_response = order_response_queue_.front();
    order_response_queue_.pop();
    try {
      auto m = order_response.SerializeToString();
      msg.body_length(std::strlen(m.c_str()));
      std::memcpy(msg.body(), m.c_str(), std::strlen(m.c_str()));
      msg.encode_header();
      std::memcpy(data_,msg.data(), msg.length());
      //std::memcpy(static_cast<void *>(msg.data()), data_, msg.length());
    } catch (std::exception& e )
    {
      std::cout << e.what();
    }
    std::cout <<"write: " << msg.body() << "\n";
    boost::asio::async_write(socket_, boost::asio::buffer(data_,msg.length()),
                   [this, self](boost::system::error_code ec, std::size_t /*length*/)
                   {
                      if (ec){
                       std::cerr << "write error:" << ec.value() << " message: " << ec.message() << "\n";
                      }
                      do_write();
                    });
 }


void do_read()
  {
    auto self(shared_from_this());
    Message msg;
    socket_.async_read_some(boost::asio::buffer(res.data(), res.header_length),
                            [this, self](boost::system::error_code ec, std::size_t length) {
                            if (!ec && res.decode_header()) {

                            std::string st(res.body());
                            boost::asio::async_read(socket_, boost::asio::buffer(res.body(),                                                                                      res.body_length()), [this](boost::system::error_code ec, std::size_t length) {
                       if (!ec) {
                           std::cout << "read " << res.body() << "\n";
                           req_.DeserializeFromChar(res.body());
                           order_request_queue_.push(req_);
                        } else {
                           if (ec) {
                               std::cerr << "read error:" << ec.value() << " message: "
                                                                        << ec.message() << "\n";
                           }
                           socket_.close();
                         }
                  });
        }
        do_read();
    });
 }

here is the io_service

class Server
{
 public:
 
  Server(boost::asio::io_service& io_service, short port,std::queue<OrderRequest> &order_request_queue,
         std::queue<OrderResponse> &order_response_queue)
      : acceptor_(io_service, tcp::endpoint(tcp::v4(), port)),
        socket_(io_service) , order_response_queue_(order_response_queue),  order_request_queue_(order_request_queue)
  {
    do_accept();
  }

 private:
  void do_accept(){
    acceptor_.async_accept(socket_,
                           [this](boost::system::error_code ec) {
                             if (!ec) {
                               std::cout << "accept connection\n";
                               std::make_shared<Session>(std::move(socket_),order_request_queue_, order_response_queue_)->start();
                             }

                             do_accept();
                           });
  }


  tcp::acceptor acceptor_;
  tcp::socket socket_;
  std::queue<OrderResponse> &order_response_queue_;
  std::queue<OrderRequest> &order_request_queue_;
};

Can we see more code? Where's the code for the thread that handles the subsequent call to `do_write()` invoked by `async_write`? — David Schwartz, Feb 18 '21 at 19:31
@DavidSchwartz if I'm understanding correctly its there the `start()` function — yaodav, Feb 18 '21 at 19:48
The `start` function creates a thread that calls `do_write` and then terminates and a thread that calls `do_read` and then terminates. Where's the thread that handles the completion of the async operations?! You don't have one? — David Schwartz, Feb 18 '21 at 20:32
@DavidSchwartz the `async_read` and `async_write` call themself in the lemda function as handler — yaodav, Feb 18 '21 at 20:36
Right. You pass them a handler for the thread that handles the completion to run. But where is the code for that thread? Where is it created? What does it do when there are no completions to handle? Your code creates two threads both of which terminate before the completion handlers are called (and don't invoke any completion handlers anyway). Where's the `executor`/`io_service` code? — David Schwartz, Feb 18 '21 at 20:38
So there's no thread to run any of the completion handlers?! How are you expecting them to run exactly? You need a thread to poll the `io_service` and run the handlers. — David Schwartz, Feb 18 '21 at 21:46

score 1 · Accepted Answer · answered Feb 18 '21 at 20:05

You should make the sample self-contained.

what is Message,
where does data_ come from,
where does res come from
why is data_ used with msg.length(),
does that include the header length,
what is the relation between req_ and res
what are the lifetimes of these.
Why are you using async with threads
where is the execution context being run, etc..

Here's a review with what you gave us:

you're using order_response_queue_ from a thread, without any locking. This really cannot be okay, since nothing on the do_write thread ever pushes onto that queue, so some other thread MUST be doing that.

To be completely honest, I think the best starting point is to remove the threads. The main reason you seem to have used them were:

to "keep the loop running", which is already a feature of the async execution context. However, you already chain from the completion handler (e.g. calling do_write again), so it's fine

to be able to "await" messages on the queue:

 while (order_response_queue_.empty()) {
 }

The usual approach is just to conditionally start the loop when the first message is pushed and let it "stop" when the queue is empty. That avoids a busy spin anyways. Just replace start with something like

 void start() {
     post(strand_, [this, self = shared_from_this()] {
         if (!order_response_queue_.empty()) {
             do_write();
         }

         do_read();
     });
 }

And have something like

 void enqueue_response(Message stuff) {
     post(strand_, [this, stuff, self = shared_from_this()] {
         order_response_queue_.push(std::move(stuff));
         if (1 == order_response_queue_.size()) { // not idle/already writing
             do_write();
         }
     });
 }

Note that every operation that uses the shared queues is synchronized on a logical strand, in case you use a multithreaded execution context.

Raw Sketch

I couldn't quite figure out the idea behind Serialize/Deserialize in relation to Message so this is probbably not working - though I feel it will give you many clues on how to simplify things:

Live On Compiler Explorer*

#include <thread>
#include <boost/asio.hpp>
#include <iostream>
#include <memory>
#include <queue>
using boost::asio::ip::tcp;

struct Message {
    struct header_t {
        // TODO add magic bytes
        std::uint32_t body_len = 0; // network byte order
    };
    enum { header_length = sizeof(header_t) };
    std::string _contents{header_length};

    size_t body_length() const {
        return length() - header_length;
    }

    void body_length(size_t n) {
        _contents.resize(n+header_length);
        encode_header();
    }

    bool decode_header() {
        assert(_contents.length() >= header_length);

        auto &header = *reinterpret_cast<header_t *>(_contents.data());
        body_length(ntohl(header.body_len));

        // TODO verify magic bytes
        return true;
    }

    void encode_header() {
        assert(_contents.length() >= header_length);

        auto &header = *reinterpret_cast<header_t *>(_contents.data());
        header.body_len = htonl(_contents.length() - header_length);
    }

    std::string SerializeToString() const { return _contents; }
    void DeserializeFromChar(char const*) { } // TODO IMPLEMENTATION

    char *data()             { return _contents.data();       } 
    char *body()             { return data() + header_length; } 
    char const *data() const { return _contents.data();       } 
    char const *body() const { return data() + header_length; } 
    size_t length() const    { return _contents.size();       } 

    bool operator<(Message const &rhs) const { return _contents < rhs._contents; }
};

struct X : std::enable_shared_from_this<X> {
    std::queue<Message> order_request_queue_, order_response_queue_;

    void start() {
        post(strand_, [this, self = shared_from_this()] {
            if (!order_response_queue_.empty()) {
                do_write();
            }

            do_read();
        });
    }

    void enqueue_response(Message stuff) {
        post(strand_, [this, stuff, self = shared_from_this()] {
            order_response_queue_.push(std::move(stuff));
            if (1 == order_response_queue_.size()) { // not idle/already writing
                do_write();
            }
        });
    }

    void do_write() {
        auto self(shared_from_this());
        auto order_response = std::move(order_response_queue_.front());
        order_response_queue_.pop();

        Message msg;
        try {
            auto m = order_response.SerializeToString();
            msg.body_length(std::strlen(m.c_str()));
            std::memcpy(msg.body(), m.c_str(), std::strlen(m.c_str()));
            msg.encode_header();

            assert(msg.length() <= data_.size());
            std::memcpy(data_.data(), msg.data(), msg.length());
            // std::memcpy(static_cast<void *>(msg.data()), data_,
            // msg.length());
        } catch (std::exception &e) {
            std::cout << e.what();
        }

        std::cout << "write: " << msg.body() << "\n";

        boost::asio::async_write(
            socket_, boost::asio::buffer(data_, msg.length()),
            [this, self](boost::system::error_code ec, std::size_t /*length*/) {
                if (ec) {
                    std::cerr << "write error:" << ec.value()
                              << " message: " << ec.message() << "\n";
                } else if (!order_response_queue_.empty()) {
                    do_write();
                }
            });
    }

    Message res, req_;
    void do_read() {
        auto self(shared_from_this());
        Message msg;
        socket_.async_read_some(
            boost::asio::buffer(res.data(), res.header_length),
            [this, self](boost::system::error_code ec, std::size_t /*length*/) {
                if (!ec && res.decode_header()) {

                    std::string st(res.body());
                    boost::asio::async_read(
                        socket_,
                        boost::asio::buffer(res.body(), res.body_length()),
                        [this](boost::system::error_code ec,
                               std::size_t /*length*/) {
                            if (!ec) {
                                std::cout << "read " << res.body() << "\n";
                                req_.DeserializeFromChar(res.body());
                                order_request_queue_.push(req_);
                            } else {
                                if (ec) {
                                    std::cerr << "read error:" << ec.value()
                                              << " message: " << ec.message()
                                              << "\n";
                                }
                                socket_.close();
                            }
                        });
                }
                do_read();
            });
    }

    using Executor = boost::asio::system_executor;
    boost::asio::strand<Executor> strand_ { Executor{} };
    tcp::socket socket_ { strand_ };
    std::array<char, 1024> data_;

    // quick & dirty connection
    X() { socket_.connect({{}, 8989}); }
};

int main() {
    auto x = std::make_shared<X>();
    x->start();

    {
        Message demo;
        std::string_view greeting{"Hello world!"};
        demo.body_length(greeting.length());
        std::copy(greeting.begin(), greeting.end(), demo.body());

        x->enqueue_response({});
    }

    boost::asio::query(boost::asio::system_executor(),
                       boost::asio::execution::context)
        .join();
}

thank you for the detailed answer, the reason for the threads on the start function is to work simultaneously on the write and the read, but from what I understood from your answer it's not passable. and my IDE doesn't recognize the system_executer (boost 1.58) — yaodav, Feb 18 '21 at 20:33
Yeah, just use io_service. That's old though. (It helps if you show your code, so I didn't have to make things up). Also, you don't need *any* threads to read and write simultaneously. That's the nature of async IO. I have many answers where I show single-threaded multi-connection full-duplex TCP servers using Asio. Specifically, I'd urge you to look at some typical full-duplex examples (either in the library chat examples or [on this site](https://stackoverflow.com/questions/65285735/asynchronous-tcp-server-using-boosts-async-write-results-in-bad-file-descriptor/65293552#65293552)) — sehe, Feb 18 '21 at 20:57

boost::async read write stuck after one read and two writes

1 Answers1

Raw Sketch