通过 boost::serialization 序列化多个 std::shared_ptr 并通过 boost::asio 发送

Question

我想通过 boost asio 将 shared_ptr 对象从客户端传输到服务器。这是我的代码：

#include <boost/archive/text_iarchive.hpp>
#include <boost/archive/text_oarchive.hpp>
#include <boost/asio.hpp>
#include <boost/serialization/export.hpp>
#include <boost/serialization/shared_ptr.hpp>
#include <chrono>
#include <iostream>
#include <sstream>
#include <string>
#include <thread>

using namespace std;

class Message {
public:
    Message() {
    }

    virtual ~Message() {
    }

    string text;

private:
    friend class boost::serialization::access;

    template <class Archive>
    void serialize(Archive &ar, const unsigned int version) {
        ar &text;
    }
};

BOOST_CLASS_EXPORT(Message)

void runClient() {
    // Give server time to startup
    this_thread::sleep_for(chrono::milliseconds(3000));

    boost::asio::ip::tcp::iostream stream("localhost", "3000");
    boost::archive::text_oarchive archive(stream);

    for (int i = 0; i < 10; i++) {
        std::shared_ptr<Message> dl = std::make_shared<Message>();
        stringstream ss;
        ss << "Hello " << i;
        dl->text = ss.str();
        archive << dl;
    }

    stream.close();
    cout << "Client shutdown" << endl;
}

void handleIncommingClientConnection(boost::asio::ip::tcp::acceptor &acceptor) {
    boost::asio::ip::tcp::iostream stream;

    acceptor.accept(*stream.rdbuf());

    boost::archive::text_iarchive archive(stream);

    while (true) {
        std::shared_ptr<Message> m;

        try {
            archive >> m;
            cout << m->text << endl;
        } catch (std::exception &ex) {
            cout << ex.what() << endl;

            if (stream.eof()) {
                cout << "eof" << endl;
                stream.close();
                cout << "Server: shutdown client handling..." << endl;
                break;
            } else
                throw ex;
        }
    }
}

void runServer() {
    boost::asio::io_service ios;
    boost::asio::ip::tcp::endpoint endpoint = boost::asio::ip::tcp::endpoint(boost::asio::ip::tcp::v4(), 3000);
    boost::asio::ip::tcp::acceptor acceptor(ios, endpoint);

    handleIncommingClientConnection(acceptor);
}

int main(int argc, char **argv) {
    thread clientThread(runClient);
    thread serverThread(runServer);

    clientThread.join();
    serverThread.join();

    return 0;
}

这是程序输出：

Hello 0
Hello 1
Hello 2
Hello 3
Hello 3
Hello 3
Hello 3
Hello 3
Client shutdown
Hello 3
Hello 3
input stream error
eof
Server: shutdown client handling...

我期待以下输出：

Hello 0
Hello 1
Hello 2
Hello 3
Hello 4
Hello 5
Hello 6
Hello 7
Client shutdown
Hello 8
Hello 9
input stream error
eof
Server: shutdown client handling...

将shared_ptr 更改为简单对象（std::shared_ptr<Message> m; 至Message m）时，一切都按预期工作。我想坚持shared_ptr。我需要改变什么？

仅序列化似乎有效：

stringstream stream;

{
    boost::archive::text_oarchive archive(stream);
    std::shared_ptr<Message> dl = std::make_shared<Message>();
    stringstream ss;
    ss << "Hello World!";
    dl->text = ss.str();
    archive << dl;
}

{
    boost::archive::text_iarchive archive(stream);
    std::shared_ptr<Message> m;
    archive >> m;
    cout << m->text << endl;
}

输出：Hello World!

Answer 1

您遇到的问题是由于object tracking 由Boost.Serialization 完成的。

根据类的使用方式和其他因素，序列化 对象可以通过内存地址进行跟踪。这可以防止同样的 对象被多次写入存档或从存档中读取。 这些存储的地址也可以用来删除创建的对象 在装载过程中因投掷而中断 例外。

文档实际上预示了这个特定问题的发生：

这可能会导致程序中出现问题[原文如此]，其中不同的副本 对象从同一个地址保存。

此外，关于对象跟踪的Class Serialization Traits 文档告诉我们，在这种特殊情况下，启用了对象跟踪：

默认跟踪特征是：

• 对于原语，track_never。
• 对于指针，track_never。即默认不跟踪地址的地址。
• 所有当前的序列化包装器，例如 boost::serialization::nvp、track_never。
• 对于所有其他类型，track_selectively。也就是说，当且仅当一个或多个 以下为真：
  • 这种类型的对象在程序中通过指针序列化的任何位置。
  • 该类已明确“导出” - 见下文。
  • 该类已在存档中明确“注册”

---

回到您的情况 - 在客户端中，由于您的循环体是如何编写的，第 5 个（及后续）Message 实例与第 4 个Message 实例分配在同一地址。您可以通过在每次迭代中检查 dl.get() 的值来验证这一点。 （在我对 coliru 的测试中，所有的实例都分配在同一个地址，所以 YMMV）。

由于对象跟踪的工作原理，所有这些 shared_ptr 实例都被认为指向同一个 Message 实例（即使您同时更改了值 - 库预计不会发生这种情况），所以额外的出现只是作为附加参考序列化。在反序列化时……老实说，这有内存泄漏和/或悬空引用问题的味道（观点，尚未对此进行详细调查）。

总结起来，所示代码的主要问题是它打破了序列化库的先决条件，即您正在序列化某个恒定状态，并且在反序列化时您重新创建相同的状态。

解决此问题的一种方法是初始化 std::vector 或 shared_ptr<Message>，其中包含要在此特定事务中传输的所有消息。同样，您将反序列化另一侧的整个向量。如果您希望有一些持久连接，则将帧添加到协议中，每个帧都包含一个包含一个消息序列的存档。

---

最少的代码修改即可完成这项工作——添加包含

#include <boost/serialization/vector.hpp>

将runClient() 改成这样：

void runClient() {
    // Give server time to startup
    this_thread::sleep_for(chrono::milliseconds(3000));

    boost::asio::ip::tcp::iostream stream("127.0.0.1", "3000");

    std::vector<std::shared_ptr<Message>> messages;
    for (int i = 0; i < 10; i++) {
        std::shared_ptr<Message> dl = std::make_shared<Message>();
        stringstream ss;
        ss << "Hello " << i;
        dl->text = ss.str();
        messages.emplace_back(dl);
    }

    boost::archive::text_oarchive archive(stream);
    archive << messages;

    stream.close();
    cout << "Client shutdown" << endl;
}

然后将handleIncommingClientConnection(...) 改成这样：

void handleIncommingClientConnection(boost::asio::ip::tcp::acceptor &acceptor) {
    boost::asio::ip::tcp::iostream stream;

    acceptor.accept(*stream.rdbuf());

    boost::archive::text_iarchive archive(stream);

    while (true) {
        try {
            std::vector<std::shared_ptr<Message>> messages;
            archive >> messages;
            for (auto const& m : messages) {
                cout << m->text << endl;
            }
        } catch (std::exception &ex) {
            cout << ex.what() << endl;

            if (stream.eof()) {
                cout << "eof" << endl;
                stream.close();
                cout << "Server: shutdown client handling..." << endl;
                break;
            } else
                throw ex;
        }
    }
}

NB：这不会增加对多帧的任何支持——客户端应该在发送一个消息向量后关闭连接，否则行为未定义。

Sample on Coliru

---

更多资源：

• boost serialization multiple objects