如何在多线程程序中使用boost :: asio正确处理fork（）？

Z时代
2024-01-10
分类：问答

我在掌握如何正确处理从以多线程方式使用Boost Asio的多线程程序创建子进程方面遇到一些麻烦。

如果我理解正确，那么在Unix世界中启动子进程的方法是先调用，fork()然后调用exec*()。另外，如果我理解正确，则调用fork()将复制所有文件描述符，依此类推，除非标记为，否则需要

在子进程中将其 关闭FD_CLOEXEC（从而在调用时被原子关闭exec*()）。

Boost

Asio需要在fork()被调用时得到通知，以便通过调用正确运行notify_fork()。但是，在多线程程序中，这会产生几个问题：

如果我理解正确的话，套接字默认是由子进程继承的。可以将它们设置为SOCK_CLOEXEC-， 但不能直接在创建时设置 *，因此，如果正在从另一个线程创建子进程，则会导致出现计时窗口。

notify_fork()要求没有其他线程调用 任何其他io_service函数，也没有 与该.I关联的任何其他I / O对象上的任何函数io_service。这似乎真的不可行-毕竟程序是多线程的，是有原因的。

如果我理解正确，则在fork()和之间进行的任何函数调用都exec*()必须是异步信号安全的（请参阅fork()文档）。没有文档说明该notify_fork()呼叫是异步信号安全的。实际上，如果我查看Boost Asio的源代码（至少在1.54版中），可能会有对pthread_mutex_lock的调用，如果我理解正确的话，这不是异步信号安全的（请参阅Signal Concepts，还有其他调用在进行中）不在白名单上）。

问题＃1我可能可以通过将子进程和套接字+文件的创建分开来解决，以便确保在正在创建和设置的套接字之间的窗口中没有子进程被创建SOCK_CLOEXEC。问题2比较棘手，我可能需要确保

所有 asio处理程序线程都已停止，执行fork，然后再次重新创建它们，这充其量是大事，而在最坏的情况下真的很糟糕（我的未决计时器是什么？？

）。问题3似乎使得完全不可能正确使用它。

…还是我“分叉”？

如果我误解了任何基本概念，请让我知道（我是在Windows编程上长大的，而不是* nix …）。

编辑：*-实际上，可以SOCK_CLOEXEC直接在Linux上创建带有套接字的套接字，此套接字自2.6.27开始可用（请参阅socket()文档）。在Windows上，WSA_FLAG_NO_HANDLE_INHERIT从Windows

7 SP 1 / Windows Server 2008 R2 SP

1开始提供相应的标志（请参阅WSASocket()文档）。OS X似乎不支持此功能。

回答：

在多线程程序中，io_service::notify_fork()在子级中调用是不安全的。但是，Boost.Asio希望基于fork()support来调用它，因为这是孩子关闭父级以前的内部文件描述符并创建新的文件描述符时。尽管Boost.Asio明确列出了调用的前提条件io_service::notify_fork()，并在期间保证了其内部组件的状态，但fork()对实现的简短了解表明std::vector::push_back()可以从空闲存储区分配内存，并且不能保证分配是异步信号-

安全。

话虽如此，一种可能值得考虑的解决方案fork()是单线程时的进程。子进程将保持单线程并执行，fork()并且exec()在父进程通过进程间通信告知执行该操作时。这种分离通过消除在执行fork()和时管理多个线程的状态的需要，简化了问题exec()。

这是一个演示此方法的完整示例，其中多线程服务器将通过UDP接收文件名，并且子进程将执行fork()并在文件名exec()上运行/usr/bin/touch。为了使该示例更具可读性，我选择使用堆栈式协程。

#include <unistd.h> // execl, fork
#include <iostream>
#include <string>
#include <boost/bind.hpp>
#include <boost/asio.hpp>
#include <boost/asio/spawn.hpp>
#include <boost/make_shared.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/thread.hpp>
/// @brief launcher receives a command from inter-process communication,
///        and will then fork, allowing the child process to return to
///        the caller.
class launcher
{
public:
  launcher(boost::asio::io_service& io_service,
           boost::asio::local::datagram_protocol::socket& socket,
           std::string& command)
    : io_service_(io_service),
      socket_(socket),
      command_(command)
  {}
  void operator()(boost::asio::yield_context yield)
  {
    std::vector<char> buffer;
    while (command_.empty())
    {
      // Wait for server to write data.
      std::cout << "launcher is waiting for data" << std::endl;
      socket_.async_receive(boost::asio::null_buffers(), yield);
      // Resize buffer and read all data.
      buffer.resize(socket_.available());
      socket_.receive(boost::asio::buffer(buffer));
      io_service_.notify_fork(boost::asio::io_service::fork_prepare);
      if (fork() == 0) // child
      {
        io_service_.notify_fork(boost::asio::io_service::fork_child);
        command_.assign(buffer.begin(), buffer.end());
      }
      else // parent
      {
        io_service_.notify_fork(boost::asio::io_service::fork_parent);
      }
    }
  }
private:
  boost::asio::io_service& io_service_;
  boost::asio::local::datagram_protocol::socket& socket_;
  std::string& command_;
};
using boost::asio::ip::udp;
/// @brief server reads filenames from UDP and then uses
///        inter-process communication to delegate forking and exec
///        to the child launcher process.
class server
{
public:
  server(boost::asio::io_service& io_service,
         boost::asio::local::datagram_protocol::socket& socket,
          short port)
    : io_service_(io_service),
      launcher_socket_(socket),
      socket_(boost::make_shared<udp::socket>(
        boost::ref(io_service), udp::endpoint(udp::v4(), port)))
  {}
  void operator()(boost::asio::yield_context yield)
  {
    udp::endpoint sender_endpoint;
    std::vector<char> buffer;
    for (;;)
    {
      std::cout << "server is waiting for data" << std::endl;
      // Wait for data to become available.
      socket_->async_receive_from(boost::asio::null_buffers(),
          sender_endpoint, yield);
      // Resize buffer and read all data.
      buffer.resize(socket_->available());
      socket_->receive_from(boost::asio::buffer(buffer), sender_endpoint);
      std::cout << "server got data: ";
      std::cout.write(&buffer[0], buffer.size());
      std::cout << std::endl;
      // Write filename to launcher.
      launcher_socket_.async_send(boost::asio::buffer(buffer), yield);
    }
  }
private:
  boost::asio::io_service& io_service_;
  boost::asio::local::datagram_protocol::socket& launcher_socket_;
  // To be used as a coroutine, server must be copyable, so make socket_
  // copyable.
  boost::shared_ptr<udp::socket> socket_;
};
int main(int argc, char* argv[])
{
  std::string filename;
  // Try/catch provides exception handling, but also allows for the lifetime
  // of the io_service and its IO objects to be controlled.
  try
  {
    if (argc != 2)
    {
      std::cerr << "Usage: <port>\n";
      return 1;
    }
    boost::thread_group threads;
    boost::asio::io_service io_service;
    // Create two connected sockets for inter-process communication.
    boost::asio::local::datagram_protocol::socket parent_socket(io_service);
    boost::asio::local::datagram_protocol::socket child_socket(io_service);
    boost::asio::local::connect_pair(parent_socket, child_socket);
    io_service.notify_fork(boost::asio::io_service::fork_prepare);
    if (fork() == 0) // child
    {
      io_service.notify_fork(boost::asio::io_service::fork_child);
      parent_socket.close();
      boost::asio::spawn(io_service,
          launcher(io_service, child_socket, filename));
    }
    else // parent
    {
      io_service.notify_fork(boost::asio::io_service::fork_parent);
      child_socket.close();
      boost::asio::spawn(io_service, 
          server(io_service, parent_socket, std::atoi(argv[1])));
      // Spawn additional threads.
      for (std::size_t i = 0; i < 3; ++i)
      {
        threads.create_thread(
          boost::bind(&boost::asio::io_service::run, &io_service));
      }
    }
    io_service.run();
    threads.join_all();
  }
  catch (std::exception& e)
  {
    std::cerr << "Exception: " << e.what() << "\n";
  }
  // Now that the io_service and IO objects have been destroyed, all internal
  // Boost.Asio file descriptors have been closed, so the execl should be
  // in a clean state.  If the filename has been set, then exec touch.
  if (!filename.empty())
  {
    std::cout << "creating file: " << filename << std::endl;
    execl("/usr/bin/touch", "touch", filename.c_str(), static_cast<char*>(0));
  }
}

1号航站楼：

$ ls a.out example.cpp $ ./a.out 12345 服务器正在等待数据启动器正在等待数据服务器获取数据：服务器正在等待数据启动器正在等待数据创建文件：服务器获取数据：b 服务器正在等待数据启动器正在等待数据创建文件：b 服务器获取数据：c 服务器正在等待数据启动器正在等待数据创建文件：c ctrl + c $ ls a.out bc example.cpp

2号航站楼：

$ nc -u 127.0.0.1 12345 a ctrl + db ctrl + dcctrl + d

以上是如何在多线程程序中使用boost :: asio正确处理fork（）？的全部内容，来源链接： utcz.com/qa/416021.html