Boost::Asio 同步客户端超时

2021-12-24 00:00:00 multithreading c++ boost boost-asio

我正在尝试使用线程作为超时控制来构建带有超时的同步 FTP 客户端代码.线程将在每个事务上启动，并在超时的情况下关闭套接字 - 这将强制同步调用返回错误.

I′m trying to build a synchronous FTP client code with timeout using a thread as the timeout control. The thread will be started on every transaction and will close the socket in case of timeout - that will force the syncronous call to return with error.

这是我的代码:

#include <cstdlib>
#include <cstring>
#include <iostream>
#include <thread>
#include <chrono>
#include <boost/asio.hpp>

#define TIMEOUT_SECONDS 5
#define MAX_MESSAGE_SIZE 4096
using boost::asio::ip::tcp;

enum { max_length = 1024 };

bool timerOn;

void socket_timer(tcp::socket& s, int seconds)
{
    std::chrono::system_clock::time_point start = std::chrono::system_clock::now();

    while (timerOn)
    {
        std::chrono::system_clock::time_point now = std::chrono::system_clock::now();
        auto interval = std::chrono::duration_cast<std::chrono::seconds>(now - start).count();

        if (interval > seconds)
            break;

        std::this_thread::sleep_for(std::chrono::milliseconds(10)); //  Not to run in 100% CPU
    }


    if (timerOn)
        s.close();
}

void start_timer(int seconds, tcp::socket& s) 
{
    timerOn = true;
    std::thread t(socket_timer, s, seconds);
    t.detach();
}

void stop_timer()
{
    timerOn = false;
}

int main(int argc, char* argv[])
{
  std::string address;

  while(address != "END")
  {
      try
      {
        boost::asio::io_service io_service;

        std::cout << "Enter FTP server address to connect or END to finish: " << std::endl;
        std::cin >> address;

        if (address == "END")
            break;

        tcp::socket s(io_service);
        tcp::resolver resolver(io_service);
        boost::asio::ip::tcp::endpoint endpoint(boost::asio::ip::address::from_string(address), 21);

        start_timer(TIMEOUT_SECONDS, s);
        boost::system::error_code ec;
        s.connect(endpoint, ec);
        stop_timer();

        if (ec)
        {
            throw std::runtime_error("Error connecting to server.");
        }

        std::cout << "Connected to " << s.remote_endpoint().address().to_string() << std::endl;

        char reply[max_length];

        start_timer(TIMEOUT_SECONDS, s);
        size_t bytes = s.receive(boost::asio::buffer(reply, MAX_MESSAGE_SIZE), 0, ec);
        stop_timer();

        if (ec)
        {
            throw std::runtime_error("Error receiving message.");
        }

        std::cout << "Received message is: ";
        std::cout.write(reply, bytes);
        std::cout << "
";

        std::cout << "Enter message: ";
        char request[max_length];
        std::cin.getline(request, max_length);
        size_t request_length = std::strlen(request);

        start_timer(TIMEOUT_SECONDS, s);
        boost::asio::write(s, boost::asio::buffer(request, request_length));
        stop_timer();

        if (ec)
        {
            throw std::runtime_error("Error sending message.");
        }
      }
      catch (std::exception& e)
      {
        std::cerr << "COMMUNICATIONS ERROR." << "
";
        std::cerr << "Exception: " << e.what() << "
";
      }
    }

   return 0;
}

我根本无法编译此代码，因为 boost 显示以下错误:

I simply cannot compile this code, as boost is showing me the following error:

1>------ Build started: Project: TestAsio, Configuration: Debug Win32 ------
1>  main.cpp
1>c:oost_1_60oostasioasic_socket.hpp(1513): error C2248: 'boost::asio::basic_io_object<IoObjectService>::basic_io_object' : cannot access private member declared in class 'boost::asio::basic_io_object<IoObjectService>'
1>          with
1>          [
1>              IoObjectService=boost::asio::stream_socket_service<boost::asio::ip::tcp>
1>          ]
1>          c:oost_1_60oostasioasic_io_object.hpp(230) : see declaration of 'boost::asio::basic_io_object<IoObjectService>::basic_io_object'
1>          with
1>          [
1>              IoObjectService=boost::asio::stream_socket_service<boost::asio::ip::tcp>
1>          ]
1>          This diagnostic occurred in the compiler generated function 'boost::asio::basic_socket<Protocol,SocketService>::basic_socket(const boost::asio::basic_socket<Protocol,SocketService> &)'
1>          with
1>          [
1>              Protocol=boost::asio::ip::tcp,
1>              SocketService=boost::asio::stream_socket_service<boost::asio::ip::tcp>
1>          ]
========== Build: 0 succeeded, 1 failed, 9 up-to-date, 0 skipped ==========

所以，我想知道两件事:

So, I wanna know about 2 things:

a) 我在代码中做错了什么?

a) What am I doing wrong in the code ?

b) 这种在并行线程上关闭套接字的方法是否可以使套接字超时?请随意发表评论.

b) Will this approach of closing the socket on a parallel thread work for timing out the socket ? Please fell free to comment it.

感谢您的帮助.

推荐答案

我已经制作了一个辅助工具来同步"执行任何 Asio 异步操作，并在此处超时，请查找 await_operation:

I've made a helper facility to do any Asio async operation "synchronously" with a timeout here, look for await_operation:

  • boost::asio + std::future - 访问关闭套接字后的违规

您应该能够为您的样本调整模式.

You should be able to adapt the pattern for your sample.

因为我想用 ftp 服务器测试它花了一段时间.

It took a while since I wanted to test this with an ftp server.

注意事项:

  • 您没有解析地址(实际上要求用户输入 IP 地址)
  • 您没有确保命令以换行符结束
  • 您没有处理任何类型的输入错误

使用我的await_operation修复这些问题和你会得到这个:

Fixing these things and using my await_operation you'd get this:

#include <cstdlib>
#include <cstring>
#include <iostream>
#include <thread>
#include <chrono>
#include <boost/asio.hpp>
#include <boost/asio/high_resolution_timer.hpp>

#define TIMEOUT std::chrono::seconds(5)
#define MAX_MESSAGE_SIZE 4096
using boost::asio::ip::tcp;

enum { max_length = 2048 };

struct Service {
    using error_code = boost::system::error_code;

    template<typename AllowTime, typename Cancel> void await_operation_ex(AllowTime const& deadline_or_duration, Cancel&& cancel) {
        using namespace boost::asio;

        ioservice.reset();
        {
            high_resolution_timer tm(ioservice, deadline_or_duration);
            tm.async_wait([&cancel](error_code ec) { if (ec != error::operation_aborted) std::forward<Cancel>(cancel)(); });
            ioservice.run_one();
        }
        ioservice.run();
    }

    template<typename AllowTime, typename ServiceObject> void await_operation(AllowTime const& deadline_or_duration, ServiceObject& so) {
        return await_operation_ex(deadline_or_duration, [&so]{ so.cancel(); });
    }

    boost::asio::io_service ioservice;
};

int main()
{
  while(true)
  {
    try
    {
      Service service;

      std::cout << "Enter FTP server address to connect or END to finish: " << std::endl;

      std::string address;
      if (std::cin >> address) {
        if (address == "END") break;
      } else {
        if (std::cin.eof())
          break;
        std::cerr << "Invalid input ignored
";
        std::cin.clear();
        std::cin.ignore(1024, '
');

        continue;
      }

      tcp::socket s(service.ioservice);
      tcp::resolver resolver(service.ioservice);

      boost::asio::async_connect(s, resolver.resolve({address, "21"}), [](boost::system::error_code ec, tcp::resolver::iterator it) {
            if (ec) throw std::runtime_error("Error connecting to server: " + ec.message());
            std::cout << "Connected to " << it->endpoint() << std::endl;
          });
      service.await_operation_ex(TIMEOUT, [&]{
            throw std::runtime_error("Error connecting to server: timeout
");
          });

      auto receive = [&] {
        boost::asio::streambuf sb;
        size_t bytes;

        boost::asio::async_read_until(s, sb, '
', [&](boost::system::error_code ec, size_t bytes_transferred) {
              if (ec) throw std::runtime_error("Error receiving message: " + ec.message());
              bytes = bytes_transferred;

              std::cout << "Received message is: " << &sb;
            });

        service.await_operation(TIMEOUT, s);
        return bytes;
      };

      receive(); // banner

      auto send = [&](std::string cmd) {
        boost::asio::async_write(s, boost::asio::buffer(cmd), [](boost::system::error_code ec, size_t /*bytes_transferred*/) {
              if (ec) throw std::runtime_error("Error sending message: " + ec.message());
            });
        service.await_operation(TIMEOUT, s);
      };

      auto ftp_command = [&](std::string cmd) {
        send(cmd + "
");
        receive(); // response
      };

      //ftp_command("USER bob");
      //ftp_command("PASS hello");

      while (true) {
        std::cout << "Enter command: ";

        std::string request;
        if (!std::getline(std::cin, request))
          break;

        ftp_command(request);
      }

    }
    catch (std::exception const& e)
    {
      std::cerr << "COMMUNICATIONS ERROR " << e.what() << "
";
    }
  }

  return 0;
}

在我的测试运行中，打印如下:

Which, in my test run, prints e.g.:

相关文章