c++ - 使用来自 streambuf 的 boost::asio::ip::tcp 将数据部分写入 TCP 套接字

Question

对于那些反对票的人请注意:这个问题与 asio 的异步方面无关(尽管异步解决方案在这里可能有意义，这是我最后提出的一个问题)。这实际上只是将 streambuf 和 ostreams 与 asio tcp 套接字包装器一起使用。示例/教程不涵盖这个特定方面( segmentation 写入调用)。

我正在为需要向外部服务器发送大量数据 (~2MB) 以响应某些事件的插件环境编写一些(希望是简单的)代码。数据需要相当及时、完整地发送，但这种情况很少见，我并不过分担心原始性能。我正在使用 Google 的 Protocol Buffers 来序列化数据。

截至目前，我有以下几乎可以工作的代码:

#include <boost/asio.hpp>

// connect to the server:
boost::asio::io_service io_service;
tcp::resolver resolver(io_service);
tcp::resolver::query query(server_address, server_port);
tcp::resolver::iterator endpoint_iterator = resolver.resolve(query);
tcp::socket socket(io_service);
boost::asio::connect(socket, endpoint_iterator);

// float_array consists of ~500,000 floats in a ProtoBuf message:
//
// message FloatArray {
//   repeated float data = 1 [packed=true];
// }

// send the serialized float_array to the server:
boost::asio::streambuf b;
std::ostream os(&b);
float_array.SerializeToOstream(&os);
boost::asio::write(socket, b);

// the TCP connection *must* now close to signal the server

这个问题是我工作的环境(多线程)认为 write() 操作花费的时间太长(它阻塞)，并终止线程。由于不允许我创建自己的线程，因此我需要将 write() 操作拆分为多个单独的写入。

我关心的是如何做到这一点。我知道我可以使用它来发送准确的字节数:

boost::asio::write(socket, b, boost::asio::transfer_exactly(65536));

但是要正确地做到这一点，我需要确切地知道 ostream 中还剩下多少字节。我注意到 b.size() 相应地减少了，所以我可以使用它。但是，为了拆分我的文章，我需要在调用此函数之间存储一些状态。

我有一个选择是在调用我的写入函数之间同时存储 streambuf b 和 ostream os，但我想知道是否有更好的方法来做到这一点。据我所知，不可能部分序列化 ProtoBuf 输出，所以我认为我一直坚持调用 float_array.SerializeToOstream()。接下来的问题是是否有直接查询 ostream 的可用字节数的正确方法，或者可能使用其他机制(boost::asio::buffer 也许吧？)。

我很高兴自己查看 boost::asio 文档，我只是在寻找有关如何继续的一些指导，因为有很多文档需要完成，我不确定哪些部分的拼图是相关的。

一个想法——是否可以使用 boost::asio 沿着这些线路创建某种单线程异步“发送器”来为我处理这种状态？例如，我能否调用某种非阻塞 write() 函数，然后进行某种回调(或经常访问的函数)来检查是否完成，然后关闭 TCP 连接？

Answer 1

虽然我无法找到一种方法来查询 ostream 以确定直接在流中“等待”的数据量，但我能够完全避免 ostreams 并将数据序列化为 char* 数组。然后可以使用 boost::asio::write() 函数以类似于旧式 C 套接字方法的方式发送:

...
tcp::socket socket(io_service);
char * buffer = new char[size];  // or a smart-ptr
float_array.SerializeToArray(static_cast<void*>(buffer, size));
void * p = static_cast<void*>(buffer);
int bytes_sent = boost::asio::write(socket, boost::asio::buffer(p, bytes_to_send);

或者，如果首选使用 boost::asio::streambuf 和 std::ostream，那么似乎可以查询 streambuf(使用 .size()) 在 ostream 被用来写入一些数据之后:

boost::asio::streambuf b;
std::ostream os(&b);
float_array.SerializeToOstream(&os);

// send a chunk of a particular size
int bytes_to_send = std::min(chunk_size, b.size());
cout << b.size() << endl;  // shows amount of remaining data
boost::asio::write(socket, b, boost::asio::transfer_exactly(bytes_to_send));
cout << b.size() << endl;  // shows a reduction in amount of remaining data

因此，如果它被多次调用(针对每个 block )，则 ostream、streambuf 和 io_service 需要保持在范围内。