几个问题。
-
您的缓冲区大小不合适。正如评论者所说，sizeof(element) 与序列化存档格式无关（这会更长，参见例如
将其更改为 buffer(bufdata) 不起作用，因为它只是一个长度为零的缓冲区（std::string 默认构造为空字符串）。考虑调整大小到适当的长度或使用boost::asio::dynamic_buffer（如果您的 Boost 版本足够新）
-
字符串流和 out_archive 的生命周期太长，导致流可能不完整。在使用结果之前，请确保已刷新流并完成存档：
std::string bufdata;
{
std::stringstream os;
{
boost::archive::text_oarchive out_archive{ os };
out_archive << snd_elem;
}
bufdata = os.str();
}
-
线
recvd_bloomdb.indexins_elem(recv_elem, i);
看起来很可疑：看起来您在迭代 recvd_bloomdb 时正在对其进行修改（这可能会导致无限循环/UB，具体取决于 indexins_elem 的工作方式）
事实上，bloom_filter 的各种副本似乎并不清楚（bloom_db 参数甚至没有在服务器代码中使用）。如果你想序列化整个数据库，为什么不让bloom_filter 本身可序列化呢？
-
如果您想在同一连接上发送多个档案，则缺少消息框架。你可以决定一个分隔符。对于文本存档，NUL 字符无效，因此可以使用：
bufdata += '\x00';
sendData(client_socket, boost::asio::buffer(bufdata));
在接收端：
std::size_t bytes_transferred = boost::asio::read_until(socket, data, '\x00', ec);
修正
这是上面提到/修复的代码：
Live On Coliru
#include <boost/archive/text_iarchive.hpp>
#include <boost/archive/text_oarchive.hpp>
#include <boost/asio.hpp>
#include <boost/serialization/serialization.hpp>
#include <iostream>
#include <mutex>
namespace {
// very sumple hack to synchronize output to std::cout
#define PRINT(e) { std::lock_guard<std::mutex> lk(s_mx); std::cout << e << std::endl; }
static std::mutex s_mx;
}
namespace ba = boost::asio;
using ba::ip::tcp;
using string_buffer = boost::asio::dynamic_string_buffer<char, std::char_traits<char>, std::allocator<char> >;
void sendData(tcp::socket& socket, boost::asio::const_buffer data);
void getData(tcp::socket& socket, string_buffer data);
typedef struct elem {
bool bit = false;
int count = 0;
uint64_t hashSum[2] = {};
uint64_t idSum = 0;
template <typename Archive> void serialize(Archive& ar, unsigned) {
ar& bit;
ar& count;
ar& hashSum[0];
ar& hashSum[1];
ar& idSum;
}
friend std::ostream& operator<<(std::ostream& os, elem const& e) {
std::ostringstream oss;
oss << std::showbase << std::boolalpha;
oss << "[" << e.bit << ","
<< std::dec << e.count << ","
<< std::hex << "{" << e.hashSum[0] << "," << e.hashSum[1] << "},"
<< e.idSum << "]";
return os << oss.str();
}
} element;
namespace mocks {
struct bloom_filter {
std::vector<elem> mock_data{
{ false, 1, { 0x7e, 0xeb }, 42 },
};
bloom_filter(...) {}
size_t size() const { return mock_data.size(); }
void indexins_elem(elem const& e, size_t i) {
assert(i < size());
mock_data.at(i) = e;
//Danger: this grows the db while it is being iterated
//mock_data.insert(mock_data.begin() + i, e);
}
void show_elemvec() const {
for (auto& el : mock_data) {
PRINT(el);
}
}
void show_elem(size_t i) const { PRINT(mock_data.at(i)); }
elem getelem(size_t i) { return mock_data.at(i); }
};
void set_elemvec(element& lhs, element const& rhs) { lhs = rhs; }
}
using mocks::bloom_filter;
using mocks::set_elemvec;
typedef boost::error_info<struct tag_errmsg, std::string> errmsg_info;
void netcom_server(unsigned short portn, bloom_filter& /*bloom_db*/) {
ba::io_service io_service;
bloom_filter recvd_bloomdb(4, "rcvbloom_db", 4096);
tcp::socket server_socket(io_service);
tcp::acceptor acceptor_server(io_service, {tcp::v4(), portn});
while (1) {
acceptor_server.accept(server_socket);
PRINT("Connection Accepted : " << server_socket.remote_endpoint());
for (auto i = 0u; i < recvd_bloomdb.size(); ++i) {
std::string bufdata;
getData(server_socket, ba::dynamic_buffer(bufdata));
element recv_elem;
try {
while (bufdata.size() && ('\x00' == bufdata.back())) {
bufdata.pop_back();
}
std::istringstream is(bufdata);
PRINT("RECV DEBUG: " << std::quoted(is.str()));
{
boost::archive::text_iarchive in_archive{ is };
in_archive >> recv_elem;
}
recvd_bloomdb.indexins_elem(recv_elem, i);
} catch (std::exception& e) {
PRINT(e.what());
}
}
PRINT("Received " << recvd_bloomdb.size() << " elements at server");
recvd_bloomdb.show_elemvec();
PRINT("Close connection " << server_socket.remote_endpoint());
server_socket.close();
}
}
std::string archive_text(elem const& e) {
std::stringstream os;
{
boost::archive::text_oarchive out_archive{ os };
out_archive << e;
}
return os.str();
}
void netcom_client(unsigned short portn, std::string serverip, bloom_filter& bloom_db) {
ba::io_service io_service;
tcp::socket client_socket(io_service);
client_socket.connect({ba::ip::address::from_string(serverip), portn});
std::this_thread::sleep_for(std::chrono::seconds(1));
for (auto i = 0u; i < bloom_db.size(); ++i) {
bloom_db.show_elem(i);
auto bufdata = archive_text(bloom_db.getelem(i));
PRINT("SEND DEBUG: " << std::quoted(bufdata));
bufdata += '\x00';
sendData(client_socket, boost::asio::buffer(bufdata));
}
PRINT("Exit netcom_client");
}
void sendData(tcp::socket& socket, boost::asio::const_buffer data) {
boost::system::error_code ec;
std::size_t bytes_transferred = boost::asio::write(socket, data, ec);
if (bytes_transferred == 0 && ec == boost::asio::error::would_block) {
PRINT(" Could not send any more");
}
PRINT(ec.message() << "bytes sent " << bytes_transferred);
}
void getData(tcp::socket& socket, string_buffer data) {
boost::system::error_code ec;
std::size_t bytes_transferred = boost::asio::read_until(socket, data, '\x00', ec);
//std::size_t bytes_transferred = boost::asio::read(socket, data, ec);
if (bytes_transferred == 0 && ec == boost::asio::error::would_block) {
PRINT("No data available");
}
PRINT(ec.message() << "bytes received " << bytes_transferred);
}
int main() {
std::thread s([] {
bloom_filter db;
netcom_server(6767, db);
});
std::thread c([] {
for (int i = 0; i < 4; ++i) {
std::this_thread::sleep_for(std::chrono::seconds(1));
bloom_filter db;
netcom_client(6767, "127.0.0.1", db);
}
});
s.join();
c.join();
}
打印类似的东西
Connection Accepted : 127.0.0.1:38126
[false,1,{0x7e,0xeb},0x2a]
SEND DEBUG: "22 serialization::archive 17 0 0 0 1 126 235 42
"
Successbytes sent 49
Exit netcom_client
Successbytes received 49
RECV DEBUG: "22 serialization::archive 17 0 0 0 1 126 235 42
"
Received 1 elements at server
[false,1,{0x7e,0xeb},0x2a]
Close connection 127.0.0.1:38126
Connection Accepted : 127.0.0.1:38128
[false,1,{0x7e,0xeb},0x2a]
SEND DEBUG: "22 serialization::archive 17 0 0 0 1 126 235 42
"
Successbytes sent 49
Exit netcom_client
Successbytes received 49
RECV DEBUG: "22 serialization::archive 17 0 0 0 1 126 235 42
"
Received 1 elements at server
[false,1,{0x7e,0xeb},0x2a]
Close connection 127.0.0.1:38128
Connection Accepted : 127.0.0.1:38130
[false,1,{0x7e,0xeb},0x2a]
SEND DEBUG: "22 serialization::archive 17 0 0 0 1 126 235 42
"
Successbytes sent 49
Exit netcom_client
Successbytes received 49
RECV DEBUG: "22 serialization::archive 17 0 0 0 1 126 235 42
"
Received 1 elements at server
[false,1,{0x7e,0xeb},0x2a]
Close connection 127.0.0.1:38130
Connection Accepted : 127.0.0.1:38132
[false,1,{0x7e,0xeb},0x2a]
SEND DEBUG: "22 serialization::archive 17 0 0 0 1 126 235 42
"
Successbytes sent 49
Exit netcom_client
Successbytes received 49
RECV DEBUG: "22 serialization::archive 17 0 0 0 1 126 235 42
"
Received 1 elements at server
[false,1,{0x7e,0xeb},0x2a]
Close connection 127.0.0.1:38132
奖金
序列化bloom_filter 类型可能很容易。另外，请注意，只要elem 是 POD 类型，它就可以按位序列化，因此可以利用这一点：
这个版本：
- 展示了如何通过位向量序列化切换到二进制存档
- 展示了如何一次序列化整个数据库
- 展示了如何禁止归档标题
- 将框架移动到
getData 和sendData，使分隔符取决于存档类型（仅NUL 在二进制流中有效）
- 大大简化了进程中的服务器/客户端代码
注意事项：
- 二进制存档不可移植（您只能期望在相同架构和版本之间共享）
- 框架当前不保留使用
read_until 读取的任何剩余数据。这通常在 TCP 中可能发生，但没有给出显示的代码。只是需要注意的事情
Live On Coliru
#include <boost/archive/binary_iarchive.hpp>
#include <boost/archive/binary_oarchive.hpp>
#include <boost/archive/text_iarchive.hpp>
#include <boost/archive/text_oarchive.hpp>
#include <boost/asio.hpp>
#include <boost/serialization/serialization.hpp>
#include <boost/serialization/binary_object.hpp>
#include <boost/serialization/vector.hpp>
#include <iostream>
#include <iomanip>
#include <mutex>
namespace {
// very sumple hack to synchronize output to std::cout
#define PRINT(e) { std::lock_guard<std::mutex> lk(s_mx); std::cout << e << std::endl; }
static std::mutex s_mx;
}
namespace ba = boost::asio;
using ba::ip::tcp;
void sendData(tcp::socket& socket, std::string const& data);
std::string getData(tcp::socket& socket);
struct element {
bool bit = false;
int count = 0;
uint64_t hashSum[2] = {};
uint64_t idSum = 0;
friend std::ostream& operator<<(std::ostream& os, element const& e) {
std::ostringstream oss;
oss << std::showbase << std::boolalpha;
oss << "[" << e.bit << ","
<< std::dec << e.count << ","
<< std::hex << "{" << e.hashSum[0] << "," << e.hashSum[1] << "},"
<< e.idSum << "]";
return os << oss.str();
}
};
BOOST_IS_BITWISE_SERIALIZABLE(element)
namespace mocks {
struct bloom_filter {
bloom_filter(...) {}
size_t size() const { return mock_data.size(); }
void indexins_elem(element const& e, size_t i) {
mock_data.at(i) = e;
//// Danger: this grows the db while it is being iterated
//assert(i <= size());
//mock_data.insert(mock_data.begin() + i, e);
}
void show_elemvec() const {
for (auto& el : mock_data) PRINT(el);
}
void show_elem(size_t i) const { PRINT(mock_data.at(i)); }
element getelem(size_t i) { return mock_data.at(i); }
private:
std::vector<element> mock_data{
{ false, 1, { 0x7e, 0xeb }, 42 },
};
friend class boost::serialization::access;
template <typename Archive> void serialize(Archive& ar, unsigned) {
ar& mock_data;
}
};
}
#define USE_TEXT
#ifdef USE_TEXT
#define ARCHIVE_ text_
static const std::string DELIMITER("\x00", 1);
struct safe_print {
std::string const& data;
friend std::ostream& operator<<(std::ostream& os, safe_print wrap) {
return os << std::quoted(wrap.data);
}
};
namespace boost { namespace serialization {
template <typename Ar>
void serialize(Ar& ar, element& e, unsigned) {
ar & e.bit & e.count & e.hashSum & e.idSum;
}
} }
#else
#define ARCHIVE_ binary_
static const std::string DELIMITER("\x00\xde\xad\xbe\xef", 5);
struct safe_print {
std::string const& data;
friend std::ostream& operator<<(std::ostream& os, safe_print wrap) {
std::ostringstream oss;
oss << std::hex << std::setfill('0');
for (uint8_t ch : wrap.data) {
oss << std::setw(2) << static_cast<int>(ch);
}
return os << oss.str();
}
};
#endif
template <typename T>
std::string archive(T const& data) {
std::stringstream os;
{
boost::archive::BOOST_PP_CAT(ARCHIVE_, oarchive) out_archive{ os, boost::archive::no_header };
out_archive << data;
}
return os.str();
}
template <typename T>
void restore(std::string const& text, T& into) {
std::istringstream is(text);
boost::archive::BOOST_PP_CAT(ARCHIVE_, iarchive) ia{is, boost::archive::no_header };
ia >> into;
}
using mocks::bloom_filter;
void netcom_server(unsigned short portn) {
ba::io_service io_service;
tcp::acceptor acceptor_server(io_service, {tcp::v4(), portn});
while (1) {
tcp::socket server_socket(io_service);
acceptor_server.accept(server_socket);
PRINT("Connection Accepted : " << server_socket.remote_endpoint());
bloom_filter recvd_bloomdb;
restore(getData(server_socket), recvd_bloomdb);
PRINT("Received " << recvd_bloomdb.size() << " elements at server");
recvd_bloomdb.show_elemvec();
}
}
void netcom_client(unsigned short portn, std::string serverip, bloom_filter& bloom_db) {
ba::io_service io_service;
tcp::socket client_socket(io_service);
client_socket.connect({ba::ip::address::from_string(serverip), portn});
sendData(client_socket, archive(bloom_db));
PRINT("Exit netcom_client");
}
void sendData(tcp::socket& socket, std::string const& data) {
PRINT("SEND DEBUG: " << safe_print{data});
std::vector<ba::const_buffers_1> frame {
ba::buffer(data),
ba::buffer(DELIMITER)
};
std::size_t bytes_transferred = boost::asio::write(socket, frame);
PRINT("bytes sent " << bytes_transferred);
}
std::string getData(tcp::socket& socket) {
std::string buffer;
std::size_t bytes_transferred = boost::asio::read_until(socket, ba::dynamic_buffer(buffer), DELIMITER);
PRINT("bytes received " << bytes_transferred);
buffer.resize(bytes_transferred);
PRINT("RECV DEBUG: " << safe_print{buffer});
return buffer;
}
int main() {
PRINT("For info: sizeof(element) = " << sizeof(element));
std::thread s([] {
bloom_filter db;
netcom_server(6767);
});
std::thread c([] {
for (int i = 0; i < 4; ++i) {
std::this_thread::sleep_for(std::chrono::seconds(1));
bloom_filter db;
netcom_client(6767, "127.0.0.1", db);
}
});
s.join();
c.join();
}
打印
For info: sizeof(element) = 32
Connection Accepted : 127.0.0.1:38134
SEND DEBUG: "0 0 0 0 1 0 0 0 0 1 2 126 235 42
"
bytes sent 34
Exit netcom_client
bytes received 34
RECV DEBUG: "0 0 0 0 1 0 0 0 0 1 2 126 235 42
"
Received 1 elements at server
[false,1,{0x7e,0xeb},0x2a]
SEND DEBUG: "0 0 0 0 1 0 0 0 0 1 2 126 235 42
"
Connection Accepted : 127.0.0.1:38136
bytes received 34
RECV DEBUG: "0 0 0 0 1 0 0 0 0 1 2 126 235 42
"
bytes sent 34
Exit netcom_client
Received 1 elements at server
[false,1,{0x7e,0xeb},0x2a]
SEND DEBUG: "0 0 0 0 1 0 0 0 0 1 2 126 235 42
"
bytes sent 34
Exit netcom_client
Connection Accepted : 127.0.0.1:38138
bytes received 34
RECV DEBUG: "0 0 0 0 1 0 0 0 0 1 2 126 235 42
"
Received 1 elements at server
[false,1,{0x7e,0xeb},0x2a]
SEND DEBUG: "0 0 0 0 1 0 0 0 0 1 2 126 235 42
"
bytes sent 34
Exit netcom_client
Connection Accepted : 127.0.0.1:38140
bytes received 34
RECV DEBUG: "0 0 0 0 1 0 0 0 0 1 2 126 235 42
"
Received 1 elements at server
[false,1,{0x7e,0xeb},0x2a]