c++ - ASIO 示例代码在它应该之前关闭套接字

Question

我需要一个使用 ASIO 的并行同步 TCP 解决方案。我正在尝试从这些示例中获取示例代码:https://github.com/jvillasante/asio-network-programming-cookbook/tree/master/src (使用 ch04 中的服务器:02_Sync_parallel_tcp_server.cpp 和 ch03 中的客户端:01_Sync_tcp_client.cpp)。

我唯一更改的是附加到文本文件的日志记录。

问题是，虽然服务器运行良好，但客户端在从服务器返回单个响应后就死机了:

libc++abi.dylib: terminating with uncaught exception of type boost::exception_detail::clone_impl<boost::exception_detail::error_info_injector<boost::system::system_error> >: shutdown: Socket is not connected

服务器代码:

#include <boost/asio.hpp>
#include <atomic>
#include <memory>
#include <thread>
#include <iostream>
#include <fstream>

using namespace boost;

class Service {
public:
  Service() = default;

  void StartHandlingClient(std::shared_ptr<asio::ip::tcp::socket> sock) {
    std::thread th{[this, sock]() { HandleClient(sock); }};
    th.detach();
  }

private:
  void HandleClient(std::shared_ptr<asio::ip::tcp::socket> sock) {
    try {
      asio::streambuf request;
      asio::read_until(*sock.get(), request, '\n');

      std::istream is(&request);
      std::string line;
      std::getline(is, line);

      std::ofstream log("logfile2.txt", std::ios_base::app | std::ios_base::out);
      log << "Request: " << line << "\n" << std::flush;

      // Emulate request processing.
      int i = 0;
      while (i != 1000000) i++;
      std::this_thread::sleep_for(std::chrono::milliseconds(500));

      // Sending response.
      std::string response = "Response\n";
      asio::write(*sock.get(), asio::buffer(response));
    } catch (std::system_error& e) {
      std::ofstream log("logfile1.txt", std::ios_base::app | std::ios_base::out);
      log << "Error occurred! Error code = " << e.code().value() << ". Message: " << e.what() << "\n" << std::flush;
    }

    // Clean up
    delete this;
  }
};

class Acceptor {
public:
  Acceptor(asio::io_service& ios, unsigned short port_num)
  : m_ios{ios}, m_acceptor{m_ios, asio::ip::tcp::endpoint{asio::ip::address_v4::any(), port_num}} {
    m_acceptor.listen();
  }

  void Accept() {
    auto sock = std::make_shared<asio::ip::tcp::socket>(m_ios);

    m_acceptor.accept(*sock.get());

    (new Service)->StartHandlingClient(sock);
  }

private:
  asio::io_service& m_ios;
  asio::ip::tcp::acceptor m_acceptor;
};

class Server {
public:
  Server() : m_stop{false} {}

  void Start(unsigned short port_num) {
    m_thread.reset(new std::thread([this, port_num]() { Run(port_num); }));
  }

  void Stop() {
    m_stop.store(true);
    m_thread->join();
  }

private:
  void Run(unsigned short port_num) {
    Acceptor acc{m_ios, port_num};

    while (!m_stop.load()) {
      acc.Accept();
    }
  }

private:
  std::unique_ptr<std::thread> m_thread;
  std::atomic<bool> m_stop;
  asio::io_service m_ios;
};

int main() {
  unsigned short port_num = 3333;

  try {
    Server srv;
    srv.Start(port_num);

    std::this_thread::sleep_for(std::chrono::seconds(60));

    srv.Stop();
  } catch (std::system_error& e) {
      std::ofstream log("logfile1.txt", std::ios_base::app | std::ios_base::out);
      log << "Error occurred! Error code = " << e.code().value() << ". Message: " << e.what() << "\n" << std::flush;
  }

  return 0;
}

客户端代码:

#include <boost/asio.hpp>
#include <iostream>
#include <fstream>

using namespace boost;

class SyncTCPClient {
public:
  SyncTCPClient(const std::string& raw_ip_address, unsigned short port_num)
  : m_ep(asio::ip::address::from_string(raw_ip_address), port_num), m_sock(m_ios) {
    m_sock.open(m_ep.protocol());
  }
  ~SyncTCPClient() { close(); }

  void connect() { m_sock.connect(m_ep); }

  std::string emulateLongComputationOp(unsigned int duration_sec) {
    std::string request = "EMULATE_LONG_COMP_OP " + std::to_string(duration_sec) + "\n";
    sendRequest(request);
    return receiveResponse();
  }

private:
  void close() {
    if (m_sock.is_open()) {
      std::ofstream log("logfile1.txt", std::ios_base::app | std::ios_base::out);
      log << "shutting down\n" << std::flush;
      m_sock.shutdown(asio::ip::tcp::socket::shutdown_both);
      log << "closing the socket\n" << std::flush;
      m_sock.close();
      log << "socket closed\n" << std::flush;
    }
  }

  void sendRequest(const std::string& request) { asio::write(m_sock, asio::buffer(request)); }

  std::string receiveResponse() {
    asio::streambuf buf;
    asio::read_until(m_sock, buf, '\n');

    std::istream input(&buf);
    std::string response;
    std::getline(input, response);

    return response;
  }

private:
  asio::io_service m_ios;
  asio::ip::tcp::endpoint m_ep;
  asio::ip::tcp::socket m_sock;
};

int main() {
  const std::string raw_ip_address = "127.0.0.1";
  const unsigned short port_num = 3333;

  try {
    SyncTCPClient client{raw_ip_address, port_num};

    // Sync connect.
    client.connect();

    std::cout << "Sending request to the server...\n";
    std::string response = client.emulateLongComputationOp(10);

    std::cout << "Response received: " << response << "\n";
  } catch (std::system_error& e) {
      std::ofstream log("logfile1.txt", std::ios_base::app | std::ios_base::out);
      log << "Error occurred! Error code = " << e.code().value() << ". Message: " << e.what() << "\n" << std::flush;
    return e.code().value();
  }

  return 0;
}

Answer 1

我没有看到太多错误，而且我无法用显示的代码重现问题。

我看到的东西:

1. 线程过程可以是静态的，因为它是无状态的(delete this 是一种代码味道)
2. 不需要分离线程(使用 boost::thread_group::join_all 会好得多)
3. 您正在从服务器和客户端写入同一个日志文件；结果未定义
4. 拼写 .store()和 .load()在 atomic<bool> 上是非惯用语
5. 拼出 *sock.get()在任何类型的智能指针上都是unforgivably un-idiomatic
6. 写作code().value() - 吞下类别 - 是一件坏事，而且 e.what()不是获取消息的方式(使用 e.code().message() )。
7. 如果你需要flush , 你不妨使用 std::endl

8. 确实没有理由在 c++14 中使用 shared_ptr:

   asio::ip::tcp::socket sock(m_ios);
   
   m_acceptor.accept(sock);
   
   std::thread([sock=std::move(sock)]() mutable { HandleClient(sock); }).detach();
   

   在 C++11 中坚持:

   auto sock = std::make_shared<asio::ip::tcp::socket>(m_ios);
   
   m_acceptor.accept(*sock);
   
   std::thread([sock] { HandleClient(*sock); }).detach();
   

   这意味着 HandleClient可以只拿ip::tcp::socket&而不是智能指针。

整合

服务器.cpp

#include <atomic>
#include <boost/asio.hpp>
#include <fstream>
#include <iostream>
#include <memory>
#include <thread>

using namespace boost;

static void HandleClient(asio::ip::tcp::socket& sock) {
    try {
        asio::streambuf buf;
        asio::read_until(sock, buf, '\n');

        std::string request;
        getline(std::istream(&buf), request);

        std::ofstream log("server.log", std::ios_base::app | std::ios_base::out);
        log << "Request: " << request << std::endl;

        // Emulate request processing.
        int i = 0;
        while (i != 1000000)
            i++;
        std::this_thread::sleep_for(std::chrono::milliseconds(500));

        // Sending response.
        std::string response = "Response\n";
        asio::write(sock, asio::buffer(response));
    } catch (std::system_error &e) {
        std::ofstream log("server.log", std::ios_base::app | std::ios_base::out);
        log << e.what() << " " << e.code() << ": " << e.code().message() << std::endl;
    }
}

class Acceptor {
  public:
    Acceptor(asio::io_service &ios, unsigned short port_num)
            : m_ios{ ios }, m_acceptor{ m_ios, asio::ip::tcp::endpoint{ asio::ip::address_v4::any(), port_num } } {
        m_acceptor.listen();
    }

    void Accept() {
        auto sock = std::make_shared<asio::ip::tcp::socket>(m_ios);

        m_acceptor.accept(*sock);

        std::thread([sock] { HandleClient(*sock); }).detach();
    }

  private:
    asio::io_service &m_ios;
    asio::ip::tcp::acceptor m_acceptor;
};

class Server {
  public:
    Server() : m_stop{ false } {}

    void Start(unsigned short port_num) {
        m_thread.reset(new std::thread([this, port_num]() { Run(port_num); }));
    }

    void Stop() {
        m_stop = true;
        m_thread->join();
    }

  private:
    void Run(unsigned short port_num) {
        Acceptor acc{ m_ios, port_num };

        while (!m_stop) {
            acc.Accept();
        }
    }

  private:
    std::unique_ptr<std::thread> m_thread;
    std::atomic<bool> m_stop;
    asio::io_service m_ios;
};

int main() {
    unsigned short port_num = 3333;

    try {
        Server srv;
        srv.Start(port_num);

        std::this_thread::sleep_for(std::chrono::seconds(60));

        srv.Stop();
    } catch (std::system_error &e) {
        std::ofstream log("server.log", std::ios_base::app | std::ios_base::out);
        log << e.what() << " " << e.code() << ": " << e.code().message() << std::endl;
    }
}

客户端.cpp

#include <boost/asio.hpp>
#include <fstream>
#include <iostream>

using namespace boost;

class SyncTCPClient {
  public:
    SyncTCPClient(const std::string &raw_ip_address, unsigned short port_num)
            : m_ep(asio::ip::address::from_string(raw_ip_address), port_num), m_sock(m_ios) {
        m_sock.open(m_ep.protocol());
    }
    ~SyncTCPClient() { close(); }

    void connect() { m_sock.connect(m_ep); }

    std::string emulateLongComputationOp(unsigned int duration_sec) {
        std::string request = "EMULATE_LONG_COMP_OP " + std::to_string(duration_sec) + "\n";
        sendRequest(request);
        return receiveResponse();
    }

  private:
    void close() {
        if (m_sock.is_open()) {
            std::ofstream log("client.log", std::ios_base::app | std::ios_base::out);
            log << "shutting down" << std::endl;
            m_sock.shutdown(asio::ip::tcp::socket::shutdown_both);
            log << "closing the socket" << std::endl;
            m_sock.close();
            log << "socket closed" << std::endl;
        }
    }

    void sendRequest(const std::string &request) { asio::write(m_sock, asio::buffer(request)); }

    std::string receiveResponse() {
        asio::streambuf buf;
        asio::read_until(m_sock, buf, '\n');

        std::string response;
        getline(std::istream(&buf), response);

        return response;
    }

  private:
    asio::io_service m_ios;
    asio::ip::tcp::endpoint m_ep;
    asio::ip::tcp::socket m_sock;
};

int main() {
    const std::string raw_ip_address = "127.0.0.1";
    const unsigned short port_num = 3333;

    try {
        SyncTCPClient client{ raw_ip_address, port_num };

        // Sync connect.
        client.connect();

        std::cout << "Sending request to the server...\n";
        std::string response = client.emulateLongComputationOp(10);

        std::cout << "Response received: " << response << std::endl;
    } catch (std::system_error &e) {
        std::ofstream log("client.log", std::ios_base::app | std::ios_base::out);
        log << e.what() << " " << e.code() << ": " << e.code().message() << std::endl;
        return e.code().value();
    }
}