c++ - boost::asio UDP "gather"操作

Question

async_receive() 的 boost::asio 文档声明它支持“一次接收到多个缓冲区”，虽然我可以编写代码，但实际上看不到它是如何(或是否)工作的。

我们遇到这样一种情况，我们的一家供应商每秒向我们发送数千个 UDP 数据包，这足以让我们在某些情况下看到“接收到的数据包被丢弃”数量激增。

如果我们真的可以在一次 async_receive() 调用中填充多个缓冲区，那将是理想的，但在测试过程中，似乎即使指定了多个缓冲区，也只会为一个数据报调用处理程序。

我已经包含了我的测试代码，抱歉它太冗长了，但我需要它能够灵活地监听多个接口(interface)/多播。

#include <boost/asio.hpp>
#include <boost/thread.hpp>
#include <boost/lexical_cast.hpp>
#include <memory>
#include <algorithm>
#include <vector>
#include <string>
#include <cstdint>

// configuration options...
std::string nic;
std::string mc;
uint16_t port = 0;
uint16_t buffer_size = 0;
uint32_t socket_buffer_size = 0;
uint32_t scat_cnt = 1;

// The raw data buffer
std::vector<uint8_t> buffer;

// The scatter/gather buffer
std::vector<boost::asio::mutable_buffer> gather_buffer;

boost::asio::io_service svc;
std::unique_ptr<boost::asio::ip::udp::socket> socket_;

size_t messages_received = 0;
size_t bytes_received = 0;

bool parse_command_line(std::vector<std::string> command_line);

void on_receive(const boost::system::error_code& ec, size_t bytes)
{
   if(!ec)
   {
      socket_->async_receive(
         gather_buffer,
         [] (const boost::system::error_code& ec, size_t bytes)
      {
         on_receive(ec, bytes);
      });

      ++messages_received;
      bytes_received += bytes;

      if(0 == messages_received % 1000)
      {
         std::cout << "Received: " << messages_received << " messages, " << bytes_received << " bytes.\n";
      }
   }
   else
   {
      std::cout << "Error: " << ec.message() << '\n';
   }
}

int main(int argc, char** argv)
{
   if(parse_command_line(std::vector<std::string>(argv, argv+argc)))
   {
      try
      {
         std::cout << "Resizing segment buffer to: " << buffer_size << std::endl;
         buffer.resize(buffer_size * scat_cnt);

         for(uint32_t x = 0; x < scat_cnt; ++x)
         {
            gather_buffer.push_back(
               boost::asio::buffer(buffer.data() + (buffer_size * x), buffer_size));
         }

         std::cout << "Setting up receiving socket." << std::endl;
         socket_.reset(new boost::asio::ip::udp::socket(svc));
         socket_->open(boost::asio::ip::udp::v4());

         socket_->set_option(boost::asio::socket_base::reuse_address(true));

         std::cout << "Binding to local NIC: " << nic << std::endl;
         socket_->bind(boost::asio::ip::udp::endpoint(boost::asio::ip::address::from_string(nic), port));

         boost::asio::socket_base::non_blocking_io no_block(true);
         socket_->io_control(no_block);

         std::cout << "Setting socket buffer size to " << socket_buffer_size << std::endl;
         boost::asio::socket_base::receive_buffer_size sock_bf_sz(socket_buffer_size);
         socket_->set_option(sock_bf_sz);

         std::cout << "Joining multicast " << mc << " on " << nic << std::endl;
         boost::asio::ip::multicast::join_group jg(boost::asio::ip::address_v4::from_string(mc), boost::asio::ip::address_v4::from_string(nic));
         socket_->set_option(jg);

         std::cout << "Listening..." << std::endl;

         socket_->async_receive(
            gather_buffer,
            [] (const boost::system::error_code& ec, size_t bytes)
         {
            on_receive(ec, bytes);
         });

         std::unique_ptr<boost::asio::io_service::work> w(new boost::asio::io_service::work(svc));

         std::cout << "Starting boost proactor..." << std::endl;
         boost::thread thread([&] () { svc.run(); });

         boost::this_thread::sleep_for(boost::chrono::seconds(60));
         w.reset();

         thread.join();
      }

      catch(boost::system::error_code& ec)
      {
         std::cout << "Boost error: " << ec.message() << '\n';
      }

      catch(...)
      {
         std::cout << "Unknown Error!\n";
      }
   }

   return 0;
}

bool parse_command_line(std::vector<std::string> command_line)
{
   for(size_t idx = 0, max_switches = command_line.size();
      idx < max_switches; ++idx)
   {
      auto& curr = command_line[idx];
      std::transform(curr.begin(), curr.end(), curr.begin(), ::tolower);

      if(curr == "-nic" && ++idx < max_switches)
      {
         nic = command_line[idx];
      }
      else if(curr == "-multicast" && ++idx < max_switches)
      {
         mc = command_line[idx];
      }
      else if(curr == "-port" && ++idx < max_switches)
      {
         port = boost::lexical_cast<uint16_t>(command_line[idx]);
      }
      else if(curr == "-bfsz" && ++idx < max_switches)
      {
         buffer_size = boost::lexical_cast<uint16_t>(command_line[idx]);
      }
      else if(curr == "-sockbfsz" && ++idx < max_switches)
      {
         socket_buffer_size = boost::lexical_cast<uint32_t>(command_line[idx]);
      }
      else if(curr == "-scattercnt" && ++idx < max_switches)
      {
         scat_cnt = boost::lexical_cast<uint32_t>(command_line[idx]);
      }
   }

   std::cout
      << "NIC:                " << nic << '\n'
      << "MC:                 " << mc << '\n'
      << "Port:               " << port << '\n'
      << "Segment Size:       " << buffer_size << '\n'
      << "Socket Buffer Size: " << socket_buffer_size << '\n'
      << "Scatter/Gather:     " << scat_cnt << std::endl;

   return 
      !nic.empty() &&
      !mc.empty() &&
      port != 0 &&
      buffer_size != 0 &&
      socket_buffer_size != 0
      ;
}

Answer 1

它会接收到多个缓冲区，但您一次最多只能接收一个数据报。 recv 就是这样工作的，我认为如果它的行为突然改变，人们会感到非常惊讶。 gather 的目的是将单个数据包拆分为多个缓冲区，如果您希望接收概念上分段的数据，这可能很有用，但它不适用于接收多个缓冲区。这基本上是 recvmsg 的包装器。

也许尝试不使用 boost::asio？