There is an example in asio, which caches the sent messages in a deque. I think when there are too many unsent messages in this deque, such as 1000, I want to process it through constbuffersequence, that is, batch sending, so the following How should the code be changed, thank you!
void deliver(const chat_message& msg)
{
bool write_in_progress = !write_msgs_.empty();
write_msgs_.push_back(msg);
if (!write_in_progress)
{
boost::asio::async_write(socket_,
boost::asio::buffer(write_msgs_.front().data(),
write_msgs_.front().length()),
boost::bind(&chat_session::handle_write, shared_from_this(),
boost::asio::placeholders::error));
}
}
void handle_write(const boost::system::error_code& error)
{
if (!error)
{
write_msgs_.pop_front();
if (!write_msgs_.empty())
{
boost::asio::async_write(socket_,
boost::asio::buffer(write_msgs_.front().data(),
write_msgs_.front().length()),
boost::bind(&chat_session::handle_write, shared_from_this(),
boost::asio::placeholders::error));
}
}
else
{
room_.leave(shared_from_this());
}
}
You can transform the deque to any container modeling the const buffer sequence concept:
std::vector<asio::const_buffer> buffers;
std::transform(
begin(write_msgs_), end(write_msgs_), back_inserter(buffers),
[](Message const& s) { return asio::buffer(s); });
async_write( //
socket_, buffers,
[this, self = shared_from_this()] //
(error_code ec, std::size_t bytes_written) {
// ...
write_msgs_.clear();
});
The transform is a force of habit here, you might prefer
std::vector<asio::const_buffer> buffers;
for (auto& s: write_msgs_)
buffers.push_back(asio::buffer(s));
Modified from this recent example How to safely write to a socket from multiple threads?:
#include <boost/asio.hpp>
#include <deque>
#include <iostream>
namespace asio = boost::asio;
using boost::system::error_code;
using asio::ip::tcp;
using Message = std::string;
class chat_session : public std::enable_shared_from_this<chat_session> {
public:
chat_session(tcp::socket socket) : socket_(std::move(socket)) {}
void start() { do_read(); }
void deliver_many(std::vector<Message> msgs) {
post(socket_.get_executor(),
[this, msgs = std::move(msgs), self = shared_from_this()] //
() mutable {
for (auto& msg : msgs) {
do_write(std::move(msg));
}
});
}
void deliver(Message msg) {
post(socket_.get_executor(),
[this, msg = std::move(msg), self = shared_from_this()] //
() mutable { do_write(std::move(msg)); });
}
private:
void do_read() {
async_read_until(
socket_, asio::dynamic_buffer(incoming_), '\0',
[this, self = shared_from_this()] //
(error_code ec, std::size_t length) {
if (!ec) {
process_message(incoming_.substr(0, length - 1));
incoming_.erase(0, length);
do_read();
} else if (ec != asio::error::eof) {
std::cerr << "Read error: " << ec.message() << std::endl;
}
});
}
void do_write(Message message)
{
write_msgs_.push_back(std::move(message)); // assumed on (implicit) strand
if (write_msgs_.size() == 1) {
write_loop();
}
}
void write_loop() {
std::cerr << "write_loop with write_msgs_.size() = " << write_msgs_.size() << std::endl;
if (write_msgs_.empty())
return;
if (write_msgs_.size() > 100) {
std::vector<asio::const_buffer> buffers;
std::transform(
begin(write_msgs_), end(write_msgs_), back_inserter(buffers),
[](Message const& s) { return asio::buffer(s); });
async_write( //
socket_, buffers,
[this, self = shared_from_this()] //
(error_code ec, std::size_t /*length*/) {
if (!ec) {
write_msgs_.clear();
write_loop();
} else if (ec != asio::error::eof) {
std::cerr << "Write error: " << ec.message() << std::endl;
}
});
} else {
async_write( //
socket_, asio::buffer(write_msgs_.front()),
[this, self = shared_from_this()] //
(error_code ec, std::size_t /*length*/) {
if (!ec) {
write_msgs_.pop_front();
write_loop();
} else if (ec != asio::error::eof) {
std::cerr << "Write error: " << ec.message() << std::endl;
}
});
}
}
void process_message(Message const& message) {
std::vector<Message> responses;
for (int i = 0; i < 200; ++i) {
responses.push_back("Response #" + std::to_string(i) + " for " +
message + "\n");
}
// dispatch/post to executor because we might be on a different thread (not in this example)
// (not in this example)
post(socket_.get_executor(),
std::bind(&chat_session::deliver_many, shared_from_this(),
std::move(responses)));
}
tcp::socket socket_;
Message incoming_;
std::deque<Message> write_msgs_;
};
class server {
public:
server(asio::any_io_executor ex, unsigned short port)
: acceptor_(ex, tcp::endpoint(tcp::v4(), port))
{
do_accept();
}
private:
void do_accept()
{
acceptor_.async_accept(
make_strand(acceptor_.get_executor()),
[this](error_code ec, tcp::socket&& s) {
if (!ec) {
std::cout << "Accepted " << s.remote_endpoint() << std::endl;
std::make_shared<chat_session>(std::move(s))->start();
}
do_accept();
});
}
tcp::acceptor acceptor_;
};
int main() {
asio::thread_pool ctx;
server s(ctx.get_executor(), 8989);
ctx.join();
}
When sending a single message from a client:
g++ -std=c++20 -O2 -Wall -pedantic -pthread main.cpp
./a.out&
sleep .5; printf 'HelloWorld\0' | nc 127.0.0.1 8989 -w1
shows e.g.:
Accepted 127.0.0.1:39538
write_loop with write_msgs_.size() = 1
Response #0 for HelloWorld
write_loop with write_msgs_.size() = 199
Response #1 for HelloWorld
Response #2 for HelloWorld
Response #3 for HelloWorld
Response #4 for HelloWorld
Response #5 for HelloWorld
Response #6 for HelloWorld
Response #7 for HelloWorld
Response #8 for HelloWorld
Response #9 for HelloWorld
Response #10 for HelloWorld
Response #11 for HelloWorld
Response #12 for HelloWorld
Response #13 for HelloWorld
Response #14 for HelloWorld
Response #15 for HelloWorld
Response #16 for HelloWorld
Response #17 for HelloWorld
Response #18 for HelloWorld
Response #19 for HelloWorld
Response #20 for HelloWorld
Response #21 for HelloWorld
Response #22 for HelloWorld
Response #23 for HelloWorld
Response #24 for HelloWorld
Response #25 for HelloWorld
Response #26 for HelloWorld
Response #27 for HelloWorld
Response #28 for HelloWorld
Response #29 for HelloWorld
Response #30 for HelloWorld
Response #31 for HelloWorld
Response #32 for HelloWorld
Response #33 for HelloWorld
Response #34 for HelloWorld
Response #35 for HelloWorld
Response #36 for HelloWorld
Response #37 for HelloWorld
Response #38 for HelloWorld
Response #39 for HelloWorld
Response #40 for HelloWorld
Response #41 for HelloWorld
Response #42 for HelloWorld
Response #43 for HelloWorld
Response #44 for HelloWorld
Response #45 for HelloWorld
Response #46 for HelloWorld
Response #47 for HelloWorld
Response #48 for HelloWorld
Response #49 for HelloWorld
Response #50 for HelloWorld
Response #51 for HelloWorld
Response #52 for HelloWorld
Response #53 for HelloWorld
Response #54 for HelloWorld
Response #55 for HelloWorld
Response #56 for HelloWorld
Response #57 for HelloWorld
Response #58 for HelloWorld
Response #59 for HelloWorld
Response #60 for HelloWorld
Response #61 for HelloWorld
Response #62 for HelloWorld
Response #63 for HelloWorld
Response #64 for HelloWorld
Response #65 for HelloWorld
Response #66 for HelloWorld
Response #67 for HelloWorld
Response #68 for HelloWorld
Response #69 for HelloWorld
Response #70 for HelloWorld
Response #71 for HelloWorld
Response #72 for HelloWorld
Response #73 for HelloWorld
Response #74 for HelloWorld
Response #75 for HelloWorld
Response #76 for HelloWorld
Response #77 for HelloWorld
Response #78 for HelloWorld
Response #79 for HelloWorld
Response #80 for HelloWorld
Response #81 for HelloWorld
Response #82 for HelloWorld
Response #83 for HelloWorld
Response #84 for HelloWorld
Response #85 for HelloWorld
Response #86 for HelloWorld
Response #87 for HelloWorld
Response #88 for HelloWorld
Response #89 for HelloWorld
Response #90 for HelloWorld
Response #91 for HelloWorld
Response #92 for HelloWorld
Response #93 for HelloWorld
Response #94 for HelloWorld
Response #95 for HelloWorld
Response #96 for HelloWorld
Response #97 for HelloWorld
Response #98 for HelloWorld
Response #99 for HelloWorld
Response #100 for HelloWorld
Response #101 for HelloWorld
Response #102 for HelloWorld
Response #103 for HelloWorld
Response #104 for HelloWorld
Response #105 for HelloWorld
Response #106 for HelloWorld
Response #107 for HelloWorld
Response #108 for HelloWorld
Response #109 for HelloWorld
Response #110 for HelloWorld
Response #111 for HelloWorld
Response #112 for HelloWorld
Response #113 for HelloWorld
Response #114 for HelloWorld
Response #115 for HelloWorld
Response #116 for HelloWorld
Response #117 for HelloWorld
Response #118 for HelloWorld
Response #119 for HelloWorld
Response #120 for HelloWorld
Response #121 for HelloWorld
Response #122 for HelloWorld
Response #123 for HelloWorld
Response #124 for HelloWorld
Response #125 for HelloWorld
Response #126 for HelloWorld
Response #127 for HelloWorld
Response #128 for HelloWorld
Response #129 for HelloWorld
Response #130 for HelloWorld
Response #131 for HelloWorld
Response #132 for HelloWorld
Response #133 for HelloWorld
Response #134 for HelloWorld
Response #135 for HelloWorld
Response #136 for HelloWorld
Response #137 for HelloWorld
Response #138 for HelloWorld
Response #139 for HelloWorld
Response #140 for HelloWorld
Response #141 for HelloWorld
Response #142 for HelloWorld
Response #143 for HelloWorld
Response #144 for HelloWorld
Response #145 for HelloWorld
Response #146 for HelloWorld
Response #147 for HelloWorld
Response #148 for HelloWorld
Response #149 for HelloWorld
Response #150 for HelloWorld
Response #151 for HelloWorld
Response #152 for HelloWorld
Response #153 for HelloWorld
Response #154 for HelloWorld
Response #155 for HelloWorld
Response #156 for HelloWorld
Response #157 for HelloWorld
Response #158 for HelloWorld
Response #159 for HelloWorld
Response #160 for HelloWorld
Response #161 for HelloWorld
Response #162 for HelloWorld
Response #163 for HelloWorld
Response #164 for HelloWorld
Response #165 for HelloWorld
Response #166 for HelloWorld
Response #167 for HelloWorld
Response #168 for HelloWorld
Response #169 for HelloWorld
Response #170 for HelloWorld
Response #171 for HelloWorld
Response #172 for HelloWorld
Response #173 for HelloWorld
Response #174 for HelloWorld
Response #175 for HelloWorld
Response #176 for HelloWorld
Response #177 for HelloWorld
Response #178 for HelloWorld
Response #179 for HelloWorld
Response #180 for HelloWorld
Response #181 for HelloWorld
Response #182 for HelloWorld
Response #183 for HelloWorld
Response #184 for HelloWorld
Response #185 for HelloWorld
Response #186 for HelloWorld
Response #187 for HelloWorld
Response #188 for HelloWorld
Response #189 for HelloWorld
Response #190 for HelloWorld
Response #191 for HelloWorld
Response #192 for HelloWorld
Response #193 for HelloWorld
Response #194 for HelloWorld
Response #195 for HelloWorld
Response #196 for HelloWorld
Response #197 for HelloWorld
Response #198 for HelloWorld
Response #199 for HelloWorld
write_loop with write_msgs_.size() = 0