Boost Asio async_read有时会在阅读时挂起，但并非总是如此

你的代码想要实现什么。缺少的位太多了。

当然，如果任务是在网络套接字上异步发送/接收流量，Asio 就是这样做的。很难看出你的代码有什么特别之处。

我建议清理更明显的问题：

• (几乎(没有错误处理(检查您的error_code -s！
• 除非你在一个有趣的平台上，否则你的格式字符串应该使用 %lu 来size_t
• 当你可以只有一个向量时，你为什么要弄乱原始数组，大小可能很糟糕？

• 如果可以使用 sizeof，则永远不要假定对象的大小：

  memcpy(&id, &trans_buffer[pos], sizeof(id));
  

• 想想看，看起来缓冲区的索引无论如何都是不安全的：

      while(pos < bytes_transferred && pos < TRANS_BUFFER_SIZE)
      {
          int id = -1;
          memcpy(&id, &buffer[pos], sizeof(id));
  

  如果例如 pos == TRANS_BUFFER_SIZE-1在这里，memcpy调用了未定义的行为...

• 为什么会发生这么多new？你正在邀请一类毛茸茸的错误进入你的代码。好像内存管理不是低级编码的致命弱点。使用值或共享指针。永远不要delete this.曾经^[1]

• 为什么有这么多重复的代码？为什么一个线程池以sender命名，而另一个thread_pool2？它包含 1 个线程。秦思思为什么将一个work项目作为原始指针，另一个作为shared_ptr？

  你可以只是：

  struct service_wrap {
      service_wrap(int threads) {
          while(threads--)
              pool.create_thread(boost::bind(&boost::asio::io_service::run, boost::ref(io_service)));
      }
      ~service_wrap() {
          io_service.post(boost::bind(&service_wrap::stop, this));
          pool.join_all();
      }
  private: // mind the initialization order!
      boost::asio::io_service io_service;
      boost::optional<boost::asio::io_service::work> work;
      boost::thread_group pool;
      void stop() { 
          work = boost::none;
      }
  };
  

  所以你可以简单地写：

  service_wrap senders(NUM_THREADS);
  service_wrap receivers(1);
  

  哇。你看到了吗？不再有出错的机会。如果修复一个池，则会自动修复另一个池。不再delete第一个项目，.reset()第二个work项目。简而言之：不再有凌乱的代码，也降低了复杂性。

• 使用例外安全锁定防护装置：

  int local_n_send = -1; // not clear naming
  {
      boost::lock_guard<boost::mutex> lk(mutex);
      n_send--;
      local_n_send = n_send;
  }
  

• broadcast的身体在write_handler()中完全重复。为什么不直接称它为：

  if(local_n_send == 0 && !done)
      broadcast();
  

• 我认为仍然存在竞争条件 - 不是对n_send本身的访问的数据竞争，但如果在释放锁后n_send达到零，则重新广播的决定可能是错误的。现在，由于broadcast()只执行异步操作，因此您可以在锁定下执行此操作并摆脱竞争条件：

  void write_handler(const error_code &ec, size_t bytes_transferred) {
      boost::lock_guard<boost::mutex> lk(mutex);
      if(!(done || --n_send))
          broadcast();
  }
  

  呜

  呜呜。现在是三行代码。代码越少，错误越少。

我的猜测是，如果你像这样努力地清理代码，你必然会找到你的线索。把它想象成你会寻找一枚丢失的结婚戒指：你不会留下一团糟。相反，你会从一个房间到另一个房间，把它全部整理干净。如果需要，先把所有东西都"扔出去"。

如果你可以让这个东西自包含/和/可重现，我什至会为你进一步调试它！

干杯

这是我在查看代码时提出的一个起点：在 Coliru 上编译

#include <boost/asio.hpp>
#include <boost/thread.hpp>
#include <boost/array.hpp>
#include <boost/make_shared.hpp>
#include <boost/ptr_container/ptr_vector.hpp>
#include <iostream>
const/*expr*/ int TRANS_TUPLE_SIZE  = 15;
const/*expr*/ int TRANS_BUFFER_SIZE = 5120 / TRANS_TUPLE_SIZE * TRANS_TUPLE_SIZE;
namespace AsioTrans
{
    using boost::system::error_code;
    using namespace boost::asio;
    typedef ip::tcp::socket             socket_t;
    typedef boost::ptr_vector<socket_t> socket_list;
    class Broadcaster
    {
    private:
        boost::array<char, TRANS_BUFFER_SIZE> trans_buffer;
        int node_id;
        int mpi_rank;
        socket_t&    dbsocket;
        socket_list& sender_sockets;
        int n_send;
        boost::mutex mutex;
        bool done;
    public:
        Broadcaster(
            socket_t& dbskt,
            socket_list& senderskts,
            int mrank,
            int id) : 
                node_id(id),
                mpi_rank(mrank),
                dbsocket(dbskt),
                sender_sockets(senderskts),
                n_send(-1),
                done(false)
        {
            // count=0;
        }
        static size_t completion_condition(const error_code& error, size_t bytes_transferred)
        {
            // TODO FIXME handler error_code here
            int remain = bytes_transferred % TRANS_TUPLE_SIZE;
            if(bytes_transferred && !remain)
            {
                return 0;
            }
            else
            {
                return TRANS_BUFFER_SIZE - bytes_transferred;
            }
        }
        void write_handler(const error_code &ec, size_t bytes_transferred)
        {
            // TODO handle errors
            // TODO check bytes_transferred
            boost::lock_guard<boost::mutex> lk(mutex);
            if(!(done || --n_send))
                broadcast();
        }
        void broadcast_handler(const error_code &ec, size_t bytes_transferred)
        {
            fprintf(stdout, "@%d, broadcast_handler: %lu bytes, mpi_size:%lu, mpi_rank: %dn", node_id, bytes_transferred, sender_sockets.size(), mpi_rank);
            if(!ec)
            {
                for(size_t pos = 0; (pos < bytes_transferred && pos < TRANS_BUFFER_SIZE); pos += TRANS_TUPLE_SIZE)
                {
                    int id = -1;
                    memcpy(&id, &trans_buffer[pos], sizeof(id));
                    if(id < 0)
                    {
                        done = true;
                        fprintf(stdout, "@%d, broadcast_handler: done!n", mpi_rank);
                        break;
                    }
                }
                {
                    boost::lock_guard<boost::mutex> lk(mutex);
                    n_send = sender_sockets.size() - 1;
                }
                for(int i = 0; size_t(i) < sender_sockets.size(); i++)
                {
                    if(i != mpi_rank)
                    {
                        async_write(
                                sender_sockets[i], 
                                buffer(trans_buffer, bytes_transferred),
                                boost::bind(&Broadcaster::write_handler, this, placeholders::error, placeholders::bytes_transferred));
                    }
                }
            }
            else
            {
                std::cerr << mpi_rank << " error: " << ec.message() << std::endl;
                delete this;
            }
        }
        void broadcast()
        {
            async_read(
                    dbsocket,
                    buffer(trans_buffer),
                    Broadcaster::completion_condition, 
                    boost::bind(&Broadcaster::broadcast_handler, this,
                        placeholders::error,
                        placeholders::bytes_transferred));
        }
    };
    struct service_wrap {
        service_wrap(int threads) {
            while(threads--)
                _pool.create_thread(boost::bind(&io_service::run, boost::ref(_service)));
        }
        ~service_wrap() {
            _service.post(boost::bind(&service_wrap::stop, this));
            _pool.join_all();
        }
        io_service& service() { return _service; }
    private: // mind the initialization order!
        io_service                        _service;
        boost::optional<io_service::work> _work;
        boost::thread_group               _pool;
        void stop() { 
            _work = boost::none;
        }
    };
    extern void AsioConnectToRemote(int, int, io_service&, socket_t&, bool);
    extern void SetupAsioConnectionsWIthOthers(io_service&, socket_list&, std::string, int, bool);
}
int main()
{
    using namespace AsioTrans;
    // there's no use in increasing #threads unless there are blocking operations
    service_wrap senders(boost::thread::hardware_concurrency()); 
    service_wrap receivers(1);
    socket_t receiver_socket(receivers.service());
    AsioConnectToRemote(5000, 1, receivers.service(), receiver_socket, true);
    socket_list send_sockets(30);
    /*hadoopNodes =*/ SetupAsioConnectionsWIthOthers(senders.service(), send_sockets, "hostFileName", 3000, false);
    int mpi_rank = send_sockets.size();
    AsioTrans::Broadcaster db_receiver(receiver_socket, send_sockets, mpi_rank, mpi_rank);
    db_receiver.broadcast();
}

^[1] 没有例外。除非无例外规则有例外。例外感知。