基于 Asio 的服务器平滑重启方案

///

#ifndef _NET_SERVE_HPP
#define _NET_SERVE_HPP
#include <mutex>
#include <thread> ///std::thread::hardware_concurrency()
#include <utility>
#ifndef _WIN32
#include <unistd.h>
#include <fcntl.h>
#endif
#include <asio.hpp>

namespace net {

class Context {
public:
  Context(const Context &) = delete;
  Context &operator=(const Context &) = delete;
  explicit Context(std::size_t iocsize) : next_(0) {
    if (iocsize == 0) {
      throw std::runtime_error("io_context size eqaul 0");
    }
    for (std::size_t i = 0; i < iocsize; i++) {
      io_context_t io_context(new asio::io_context);
      work_t work(new io_work_t(asio::make_work_guard(*io_context)));
      io_contexts_.push_back(io_context);
      works_.push_back(work);
    }
  }
  void Execute() {
    std::vector<std::shared_ptr<std::thread>> threads;
    for (auto &ctx : io_contexts_) {
      std::shared_ptr<std::thread> thread(
          new std::thread([ctx]() { ctx->run(); }));
      threads.push_back(thread);
    }
    for (auto &t : threads) {
      t->join();
    }
  }
  void Exit() {
    for (auto &ioc : io_contexts_) {
      ioc->stop();
    }
  }
  void Clear() {
    //
    works_.clear();
  }

  asio::io_context &Next() {
    auto &ioc = *io_contexts_[next_];
    ++next_;
    if (next_ == io_contexts_.size()) {
      next_ = 0;
    }
    return ioc;
  }

private:
  using io_context_t = std::shared_ptr<asio::io_context>;
  using io_work_t = asio::executor_work_guard<asio::io_context::executor_type>;
  using work_t = std::shared_ptr<io_work_t>;
  std::vector<io_context_t> io_contexts_;
  std::vector<work_t> works_;
  std::size_t next_;
};

template <typename ServeVars, typename ServeSession> class Serve {
public:
  Serve(std::size_t n) : context_(n), acceptor_(context_.Next()) {}
  void SetVars(ServeVars &&vars) {
    ///
    vars_ = std::move(vars);
  }
  void Acceptorclose(std::error_code &ec) {
    /// close acceptor
    acceptor_.close(ec);
  }
  void DelayExit() {
    std::error_code ec;
    acceptor_.close(ec);
    context_.Clear();
  }
  void Exit() { context_.Exit(); }
  int ListenAndServe(const std::string &addr, int port) {
    std::error_code ec;
    asio::ip::tcp::endpoint ep(asio::ip::make_address(addr, ec),
                               static_cast<unsigned int>(port));
    if (ec) {
      fprintf(stderr, "make_address: %s\n", ec.message().c_str());
      return -1;
    }
    acceptor_.open(ep.protocol());
    acceptor_.set_option(asio::ip::tcp::acceptor::reuse_address(true));
/*
When the server needs to be supported for graceful restart, the old sockets
need to be closed, and when invoke the fork and exec* syscall, the file
descriptor is inherited, so the FD_CLOEXEC flag should be set so that the
listener socket can be closed in time.
*/
#ifndef _WIN32
    int oflag = fcntl(acceptor_.native_handle(), F_GETFD, 0);
    if (oflag >= 0) {
      fcntl(acceptor_.native_handle(), F_SETFD, oflag | FD_CLOEXEC);
    }
#endif
    for (int i = 0; i < 10; i++) {
      acceptor_.bind(ep, ec);
      if (!ec) {
        break;
      }
      fprintf(stderr, "bind: %s:%d %s\n", addr.c_str(), port,
              ec.message().c_str());
    }
    if (ec) {
      fprintf(stderr, "bind address failed, exit\n");
      return 1;
    }

    acceptor_.listen();
    Accept();
    context_.Execute();
    return 0;
  }

private:
  void Accept() {
    acceptor_.async_accept(
        context_.Next(),
        [this](std::error_code ec, asio::ip::tcp::socket socket) {
          if (ec) {
            if (socket.is_open()) {
            }
            return;
          }
          std::make_shared<ServeSession>(std::move(socket), vars_)->run();
          Accept();
        });
  }

  Context context_;
  asio::ip::tcp::acceptor acceptor_;
  ServeVars vars_;
};

}; // namespace net

#endif