使用 Boost 发送序列化结构但得到输入流错误

Question

这是here的后续问题。

我正在解决来自2016 Marathon of Parallel Programming 的第一个挑战。这个问题与第一个挑战有关：字符串解析。

如果你看了here的问题集，显然第5步是可并行化的，也就是第3页的[Try all derivations recursively.]。

所以我想通过使用 ZeroMQ 和 Boost 在下面实现这个解决方案来并行化解决方案（序列化结构 res_package，其中包含工作线程的作业和主线程的结果）：

main 中的 PUSH 套接字组用于分发任务。 main 中的 PULL 套接字组从工作线程收集结果。最后，PUB 套接字组用于向工作人员发送终止信号。

main 和worker 之间有一个同步步骤，否则main 中的PUSH 套接字会向第一个连接的工作人员发送大量作业。我引用我的帖子：

所以主线程所做的是：每次将同步消息的#worker_num 及其 PUSH 端点推送到工作线程，然后从其 PULL 端点读取确认消息。如果主线程检索到确认消息的#worker_num，则同步完成。 worker 同步消息的格式为：字符串中的工作线程 ID。所以线程 0 会将字符串中的 0 传递回主线程。

如果主线程接收到一个有意义的结果，即Eval，其第二个字段为true（表示该字符串被语法接受），则主线程将发布终止信号。待所有工作线程返回确认，即struct res_package中的字段bool exit_confirmed后，主线程加入工作线程并打印最终结果。

问题是，我从 Boost 收到运行时错误。我不知道发生了什么：

# ./spec < ./spec.in
main() : creating thread, 0
thread 0 receives: sync
to_string 0
thread 0 sends: 0, with size: 1
thread 0 sync done
pass 0 to if_sync_done
main thread receives sync msg from thread 0
sync done in main thread
456Dynamic exception type: boost::archive::archive_exception
std::exception::what: input stream error

此异常是由第 912 行（来自 GDB 回溯）引起的。所以我猜这个问题是关于悬空指针，或者在第 909 行接收到的以某种方式被截断。但我不知道如何更进一步。

要构建我的项目，您需要一些文件：spec.cc、spec.hh、Makefile

您可以从here 下载原始项目。有spec.in和judge.in供测试。

要在 Ubuntu 上安装依赖项，请运行：

apt-get install -y libzmqpp3 libzmqpp-dev libzmq5 libzmq5-dbg libboost-all-dev build-essential g++

一个重要的数据结构是：

struct res_package
{
  int i;
  Set <Stack> ls; // from worker thread to main thread
  Stack s; // from main thread to worker thread, job to process
  bool if_accepted;
  bool exit_confirmed;
  Eval res;
  bool set_nonempty;

// striped.....

  template <typename Archive>
  void serialize(Archive& ar, const unsigned int version)
  {
    ar & i;
    ar & if_accepted;
    ar & exit_confirmed;
    ar & s;
    ar & ls;
    ar & res;
    ar & set_nonempty;
  }
}

工作线程接收Stack s 并查看该字符串是否可以被规则s 接受。如果是这样，它将if_accepted 设置为true，然后发回Eval res。如果没有并且找到多个可能的规则，它会发回一组所有可能的规则ls，并将bool set_nonempty 设置为true，以指示主要分发这些规则。

如果您感到困惑，请发表评论。

Answer 1

您没有在帖子中包含序列化代码，但我在 spec.cc 中找到了它：

{
    std::ostringstream obuffer;
    boost::archive::text_oarchive oarchive(obuffer);

    tmp.exit_confirmed = true;
    oarchive & tmp;
    std::string req_str(obuffer.str());
    zmq::message_t res_msg(req_str.size());
    memcpy((void *)res_msg.data(), req_str.data(), req_str.size());
    sendres_socket.send(res_msg);
}

在这里，您可以在存档关闭之前发送流。尝试显式关闭它，或限制生命周期：

{
    std::ostringstream obuffer;
    {
        boost::archive::text_oarchive oarchive(obuffer);

        tmp.exit_confirmed = true;
        oarchive & tmp;
    }
    std::string req_str(obuffer.str());
    zmq::message_t res_msg(req_str.size());
    memcpy((void *)res_msg.data(), req_str.data(), req_str.size());
    sendres_socket.send(res_msg);
}