c++ - 使用 Boost 发送序列化结构但出现输入流错误

Question

这是 here 的后续问题。

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

如果你阅读 here 的问题集，显然第 5 步是可并行化的，即第 3 页的 [递归尝试所有推导。]。

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

main中的PUSH socket组是用来分发任务的。 main 中的 PULL 套接字组从工作线程收集结果。最后，PUB 套接字组用于向 worker 发送 kill 信号。

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

So what main thread does is: pushing #worker_num of sync msgs with its PUSH endpoint to worker threads each time and then reads confirmation msg from its PULL endpoint. If main thread retrieves #worker_num of confirmation msgs, then sync done. Format of the sync msg from worker is: the worker thread's ID in a string. So thread 0 would pass a 0 in string back to main thread.

如果主线程收到一个有意义的结果，即 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 以指示 main 分发这些规则。

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

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);
}