ASIO example code closing socket before it should

Question

I need a parallel synchronous TCP solution using ASIO. I'm trying to get the example code from these examples working: https://github.com/jvillasante/asio-network-programming-cookbook/tree/master/src (using the server in ch04: 02_Sync_parallel_tcp_server.cpp and the client in ch03: 01_Sync_tcp_client.cpp).

The only thing I changed is the logging to append to text files.

The problem is that while the server runs fine, the client dies after returning a single response from the server:

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

Code for the server:

#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;
}

Code for the client:

#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

I don't see a lot wrong, and I cannot reproduce the problem with the code shown.

Things I do see:

1. the thread procedure could be a static because it's stateless (delete this is a code smell)
2. the thread needn't be detached (using boost::thread_group::join_all would be much better)
3. you were writing to the same logfile from server as well as client; results are undefined
4. spelling .store() and .load() on an atomic<bool> is un-idiomatic
5. spelling out *sock.get() on any kind of smart pointer is unforgivably un-idiomatic
6. writing code().value() - swallowing the category - is a BAD thing to do, and e.what() is NOT the way to get the message (use e.code().message()).
7. If you need flush, you might as well use std::endl

8. There's really no reason to use a shared_ptr in c++14:

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

   In C++11 stick to:

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

   This means HandleClient can just take a ip::tcp::socket& instead of a smart pointer.

INTEGRATING

Server.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;
    }
}

Client.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();
    }
}