Allow to stop server and restart server; allow to run loop in multiple threads.

webcc/connection.cc

@@ -6,18 +6,15 @@
 
 #include "webcc/connection_pool.h"
 #include "webcc/logger.h"
-#include "webcc/server.h"
 
 using boost::asio::ip::tcp;
 
 namespace webcc {
 
 Connection::Connection(tcp::socket socket, ConnectionPool* pool,
-                       Server* server)
-    : socket_(std::move(socket)),
-      pool_(pool),
-      buffer_(kBufferSize),
-      server_(server) {
+                       Queue<ConnectionPtr>* queue)
+    : socket_(std::move(socket)), pool_(pool), queue_(queue),
+      buffer_(kBufferSize) {
 }
 
 void Connection::Start() {
@@ -122,9 +119,9 @@ void Connection::OnRead(boost::system::error_code ec, std::size_t length) {
 
   LOG_VERB("HTTP request:\n%s", request_->Dump().c_str());
 
-  // Enqueue this connection.
-  // Some worker thread will handle it later.
-  server_->Enqueue(shared_from_this());
+  // Enqueue this connection once the request has been read.
+  // Some worker thread will handle the request later.
+  queue_->Push(shared_from_this());
 }
 
 void Connection::DoWrite() {

webcc/connection.h

@@ -8,19 +8,23 @@
 #include "boost/asio/ip/tcp.hpp"
 
 #include "webcc/globals.h"
+#include "webcc/queue.h"
 #include "webcc/request.h"
 #include "webcc/request_parser.h"
 #include "webcc/response.h"
 
 namespace webcc {
 
+class Connection;
 class ConnectionPool;
 class Server;
 
+using ConnectionPtr = std::shared_ptr<Connection>;
+
 class Connection : public std::enable_shared_from_this<Connection> {
 public:
   Connection(boost::asio::ip::tcp::socket socket, ConnectionPool* pool,
-             Server* server);
+             Queue<ConnectionPtr>* queue);
 
   ~Connection() = default;
 
@@ -61,15 +65,15 @@ private:
   // The socket for the connection.
   boost::asio::ip::tcp::socket socket_;
 
-  // The pool for this connection.
+  // The connection pool.
   ConnectionPool* pool_;
 
+  // The connection queue.
+  Queue<ConnectionPtr>* queue_;
+
   // The buffer for incoming data.
   std::vector<char> buffer_;
 
-  // The server.
-  Server* server_;
-
   // The incoming request.
   RequestPtr request_;
 
@@ -80,8 +84,6 @@ private:
   ResponsePtr response_;
 };
 
-using ConnectionPtr = std::shared_ptr<Connection>;
-
 }  // namespace webcc
 
 #endif  // WEBCC_CONNECTION_H_

webcc/connection_pool.cc

@@ -6,17 +6,32 @@ namespace webcc {
 
 void ConnectionPool::Start(ConnectionPtr c) {
   LOG_VERB("Starting connection...");
-  connections_.insert(c);
+
+  {
+    // Lock the container only.
+    std::lock_guard<std::mutex> lock(mutex_);
+    connections_.insert(c);
+  }
+
   c->Start();
 }
 
 void ConnectionPool::Close(ConnectionPtr c) {
   LOG_VERB("Closing connection...");
-  connections_.erase(c);
+
+  {
+    // Lock the container only.
+    std::lock_guard<std::mutex> lock(mutex_);
+    connections_.erase(c);
+  }
+
   c->Close();
 }
 
-void ConnectionPool::CloseAll() {
+void ConnectionPool::Clear() {
+  // Lock all since we are going to stop anyway.
+  std::lock_guard<std::mutex> lock(mutex_);
+
   if (!connections_.empty()) {
     LOG_VERB("Closing all (%u) connections...", connections_.size());
     for (auto& c : connections_) {

webcc/connection_pool.h

@@ -1,6 +1,7 @@
 #ifndef WEBCC_CONNECTION_POOL_H_
 #define WEBCC_CONNECTION_POOL_H_
 
+#include <mutex>
 #include <set>
 
 #include "webcc/connection.h"
@@ -14,17 +15,24 @@ public:
   ConnectionPool(const ConnectionPool&) = delete;
   ConnectionPool& operator=(const ConnectionPool&) = delete;
 
-  // Add a connection to the pool and start it.
+  // Add the connection and start to read the request from it.
+  // Called when a new connection has just been accepted.
   void Start(ConnectionPtr c);
 
-  // Close a connection.
+  // Close the connection.
+  // Called when the response of the connection has been sent back.
   void Close(ConnectionPtr c);
 
   // Close all pending connections.
-  void CloseAll();
+  // Called when the server is about to stop.
+  void Clear();
 
 private:
   std::set<ConnectionPtr> connections_;
+
+  // Mutex is necessary if the loop is running in multiple threads.
+  // See Server::Run().
+  std::mutex mutex_;
 };
 
 }  // namespace webcc

webcc/globals.h

@@ -178,10 +178,11 @@ enum class ContentEncoding {
 
 // -----------------------------------------------------------------------------
 
-// Error (or exception) for the client.
-class Error {
- public:
+// Error or exception (for client only).
+class Error : public std::exception {
+public:
   enum Code {
+    kUnknownError = -1,
     kOK = 0,
     kSyntaxError,
     kResolveError,
@@ -193,11 +194,16 @@ class Error {
     kDataError,
   };
 
- public:
+public:
   Error(Code code = kOK, const std::string& message = "")
       : code_(code), message_(message), timeout_(false) {
   }
 
+  // Note that `noexcept` is required by GCC.
+  const char* what() const WEBCC_NOEXCEPT override{
+    return message_.c_str();
+  }
+
   Code code() const {
     return code_;
   }
@@ -223,7 +229,7 @@ class Error {
     return code_ != kOK;
   }
 
- private:
+private:
   Code code_;
   std::string message_;
   bool timeout_;

webcc/request_builder.cc

@@ -22,9 +22,9 @@ RequestPtr RequestBuilder::operator()() {
   }
 
   // If no Keep-Alive, explicitly set `Connection` to "Close".
-  if (!keep_alive_) {
+  //if (!keep_alive_) {
     request->SetHeader(headers::kConnection, "Close");
-  }
+  //}
 
   if (body_) {
     request->SetContentType(media_type_, charset_);

webcc/response_builder.h

@@ -30,11 +30,34 @@ public:
   // consistency and simplicity.
 
   // Some shortcuts for different status codes:
-  ResponseBuilder& OK() { return Code(Status::kOK); }
-  ResponseBuilder& Created() { return Code(Status::kCreated); }
-  ResponseBuilder& BadRequest() { return Code(Status::kBadRequest); }
-  ResponseBuilder& NotFound() { return Code(Status::kNotFound); }
-  ResponseBuilder& NotImplemented() { return Code(Status::kNotImplemented); }
+
+  ResponseBuilder& OK() {
+    return Code(Status::kOK);
+  }
+
+  ResponseBuilder& Created() {
+    return Code(Status::kCreated);
+  }
+
+  ResponseBuilder& BadRequest() {
+    return Code(Status::kBadRequest);
+  }
+
+  ResponseBuilder& NotFound() {
+    return Code(Status::kNotFound);
+  }
+
+  ResponseBuilder& InternalServerError() {
+    return Code(Status::kInternalServerError);
+  }
+
+  ResponseBuilder& NotImplemented() {
+    return Code(Status::kNotImplemented);
+  }
+
+  ResponseBuilder& ServiceUnavailable() {
+    return Code(Status::kServiceUnavailable);
+  }
 
   ResponseBuilder& Code(Status code) {
     code_ = code;

webcc/server.cc

@@ -21,42 +21,9 @@ using tcp = boost::asio::ip::tcp;
 namespace webcc {
 
 Server::Server(std::uint16_t port, const Path& doc_root)
-    : acceptor_(io_context_), signals_(io_context_), doc_root_(doc_root) {
-  RegisterSignals();
-
-  boost::system::error_code ec;
-  tcp::endpoint endpoint(tcp::v4(), port);
-
-  // Open the acceptor.
-  acceptor_.open(endpoint.protocol(), ec);
-  if (ec) {
-    LOG_ERRO("Acceptor open error (%s).", ec.message().c_str());
-    return;
-  }
-
-  // Set option SO_REUSEADDR on.
-  // When SO_REUSEADDR is set, multiple servers can listen on the same port.
-  // This is necessary for restarting the server on the same port.
-  // More details:
-  // - https://stackoverflow.com/a/3233022
-  // - http://www.andy-pearce.com/blog/posts/2013/Feb/so_reuseaddr-on-windows/
-  acceptor_.set_option(tcp::acceptor::reuse_address(true));
-
-  // Bind to the server address.
-  acceptor_.bind(endpoint, ec);
-  if (ec) {
-    LOG_ERRO("Acceptor bind error (%s).", ec.message().c_str());
-    return;
-  }
-
-  // Start listening for connections.
-  // After listen, the client is able to connect to the server even the server
-  // has not started to accept the connection yet.
-  acceptor_.listen(boost::asio::socket_base::max_listen_connections, ec);
-  if (ec) {
-    LOG_ERRO("Acceptor listen error (%s).", ec.message().c_str());
-    return;
-  }
+    : port_(port), doc_root_(doc_root), running_(false),
+      acceptor_(io_context_), signals_(io_context_) {
+  AddSignals();
 }
 
 bool Server::Route(const std::string& url, ViewPtr view,
@@ -88,50 +55,73 @@ bool Server::Route(const UrlRegex& regex_url, ViewPtr view,
   return true;
 }
 
-void Server::Start(std::size_t workers) {
+void Server::Run(std::size_t workers, std::size_t loops) {
   assert(workers > 0);
-  assert(worker_threads_.empty());
 
-  if (!acceptor_.is_open()) {
-    LOG_ERRO("Server is NOT going to run.");
-    return;
+  {
+    std::lock_guard<std::mutex> lock(state_mutex_);
+
+    assert(worker_threads_.empty());
+
+    if (IsRunning()) {
+      LOG_WARN("Server is already running.");
+      return;
+    }
+
+    running_ = true;
+    io_context_.restart();
+
+    if (!Listen(port_)) {
+      LOG_ERRO("Server is NOT going to run.");
+      return;
+    }
+
+    LOG_INFO("Server is going to run...");
+
+    AsyncWaitSignals();
+
+    AsyncAccept();
+
+    // Create worker threads.
+    for (std::size_t i = 0; i < workers; ++i) {
+      worker_threads_.emplace_back(std::bind(&Server::WorkerRoutine, this));
+    }
   }
 
-  LOG_INFO("Server is going to run...");
-
-  DoAwaitStop();
-
-  DoAccept();
-
-  // Create worker threads.
-  for (std::size_t i = 0; i < workers; ++i) {
-    worker_threads_.emplace_back(std::bind(&Server::WorkerRoutine, this));
-  }
-
-  // Run the loop.
+  // Start the event loop.
   // The io_context::run() call will block until all asynchronous operations
   // have finished. While the server is running, there is always at least one
   // asynchronous operation outstanding: the asynchronous accept call waiting
   // for new incoming connections.
-  io_context_.run();
+
+  LOG_INFO("Loop is running in %u thread(s).", loops);
+
+  if (loops == 1) {
+    // Just run the loop in the current thread.
+    io_context_.run();
+  } else {
+    std::vector<std::thread> loop_threads;
+    for (std::size_t i = 0; i < loops; ++i) {
+      loop_threads.emplace_back(&boost::asio::io_context::run, &io_context_);
+    }
+    // Join the threads for blocking.
+    for (std::size_t i = 0; i < loops; ++i) {
+      loop_threads[i].join();
+    }
+  }
 }
 
 void Server::Stop() {
-  // Stop listener.
-  acceptor_.close();
+  std::lock_guard<std::mutex> lock(state_mutex_);
 
-  // Stop worker threads.
-  StopWorkers();
-
-  // Close all pending connections.
-  pool_.CloseAll();
+  DoStop();
 }
 
-void Server::Enqueue(ConnectionPtr connection) {
-  queue_.Push(connection);
+bool Server::IsRunning() const {
+  return running_ && !io_context_.stopped();
 }
 
-void Server::RegisterSignals() {
+void Server::AddSignals() {
   signals_.add(SIGINT);  // Ctrl+C
   signals_.add(SIGTERM);
 
@@ -140,7 +130,58 @@ void Server::RegisterSignals() {
 #endif
 }
 
-void Server::DoAccept() {
+void Server::AsyncWaitSignals() {
+  signals_.async_wait(
+      [this](boost::system::error_code, int signo) {
+        // The server is stopped by canceling all outstanding asynchronous
+        // operations. Once all operations have finished the io_context::run()
+        // call will exit.
+        LOG_INFO("On signal %d, stopping the server...", signo);
+
+        DoStop();
+      });
+}
+
+bool Server::Listen(std::uint16_t port) {
+  boost::system::error_code ec;
+
+  tcp::endpoint endpoint(tcp::v4(), port);
+
+  // Open the acceptor.
+  acceptor_.open(endpoint.protocol(), ec);
+  if (ec) {
+    LOG_ERRO("Acceptor open error (%s).", ec.message().c_str());
+    return false;
+  }
+
+  // Set option SO_REUSEADDR on.
+  // When SO_REUSEADDR is set, multiple servers can listen on the same port.
+  // This is necessary for restarting the server on the same port.
+  // More details:
+  // - https://stackoverflow.com/a/3233022
+  // - http://www.andy-pearce.com/blog/posts/2013/Feb/so_reuseaddr-on-windows/
+  acceptor_.set_option(tcp::acceptor::reuse_address(true));
+
+  // Bind to the server address.
+  acceptor_.bind(endpoint, ec);
+  if (ec) {
+    LOG_ERRO("Acceptor bind error (%s).", ec.message().c_str());
+    return false;
+  }
+
+  // Start listening for connections.
+  // After listen, the client is able to connect to the server even the server
+  // has not started to accept the connection yet.
+  acceptor_.listen(boost::asio::socket_base::max_listen_connections, ec);
+  if (ec) {
+    LOG_ERRO("Acceptor listen error (%s).", ec.message().c_str());
+    return false;
+  }
+
+  return true;
+}
+
+void Server::AsyncAccept() {
   acceptor_.async_accept(
       [this](boost::system::error_code ec, tcp::socket socket) {
         // Check whether the server was stopped by a signal before this
@@ -153,25 +194,33 @@ void Server::DoAccept() {
           LOG_INFO("Accepted a connection.");
 
           auto connection = std::make_shared<Connection>(
-              std::move(socket), &pool_, this);
+              std::move(socket), &pool_, &queue_);
 
           pool_.Start(connection);
         }
 
-        DoAccept();
+        AsyncAccept();
       });
 }
 
-void Server::DoAwaitStop() {
-  signals_.async_wait(
-      [this](boost::system::error_code, int signo) {
-        // The server is stopped by canceling all outstanding asynchronous
-        // operations. Once all operations have finished the io_context::run()
-        // call will exit.
-        LOG_INFO("On signal %d, stopping the server...", signo);
+void Server::DoStop() {
+  // Stop accepting new connections.
+  acceptor_.close();
 
-        Stop();
-      });
+  // Stop worker threads.
+  // This might take some time if the threads are still processing.
+  StopWorkers();
+
+  // Close all pending connections.
+  pool_.Clear();
+
+  // Finally, stop the event processing loop.
+  // This function does not block, but instead simply signals the io_context to
+  // stop. All invocations of its run() or run_one() member functions should
+  // return as soon as possible.
+  io_context_.stop();
+
+  running_ = false;
 }
 
 void Server::WorkerRoutine() {
@@ -186,7 +235,7 @@ void Server::WorkerRoutine() {
       // For stopping next worker.
       queue_.Push(ConnectionPtr());
 
-      // Stop the worker.
+      // Stop this worker.
       break;
     }
 
@@ -198,19 +247,27 @@ void Server::StopWorkers() {
   LOG_INFO("Stopping workers...");
 
   // Clear pending connections.
-  // The connections will be closed later (see Server::DoAwaitStop).
+  // The connections will be closed later.
   LOG_INFO("Clear pending connections...");
   queue_.Clear();
 
   // Enqueue a null connection to trigger the first worker to stop.
   queue_.Push(ConnectionPtr());
 
+  // Wait for worker threads to finish.
   for (auto& t : worker_threads_) {
     if (t.joinable()) {
       t.join();
     }
   }
 
+  // Cleanup worker threads.
+  worker_threads_.clear();
+
+  // Clear the queue because it has a remaining null connection pushed by the
+  // last worker thread.
+  queue_.Clear();
+
   LOG_INFO("All workers have been stopped.");
 }
 

webcc/server.h

@@ -21,7 +21,7 @@ class Server {
 public:
   explicit Server(std::uint16_t port, const Path& doc_root = {});
 
-  virtual ~Server() = default;
+  ~Server() = default;
 
   Server(const Server&) = delete;
   Server& operator=(const Server&) = delete;
@@ -37,24 +37,43 @@ public:
   bool Route(const UrlRegex& regex_url, ViewPtr view,
              const Strings& methods = { "GET" });
 
-  // Start the server with a given number of worker threads.
-  void Start(std::size_t workers = 1);
+  // Start and run the server.
+  // This method is blocking so will not return until Stop() is called (from
+  // another thread) or a signal like SIGINT is caught.
+  // When the request of a connection has been read, the connection is put into
+  // a queue waiting for some worker thread to process. Normally, the more
+  // |workers| you have, the more concurrency you gain (the concurrency also
+  // depends on the number of CPU cores). The worker thread pops connections
+  // from the queue one by one, prepares the response by the user provided View,
+  // then sends it back to the client.
+  // Meanwhile, the (event) loop, i.e., io_context, is also running in a number
+  // (|loops|) of threads. Normally, one thread for the loop is good enough, but
+  // it could be more than that.
+  void Run(std::size_t workers = 1, std::size_t loops = 1);
 
   // Stop the server.
+  // This should be called from another thread since the Run() is blocking.
   void Stop();
 
-  // Put the connection into the queue.
-  void Enqueue(ConnectionPtr connection);
+  // Is the server running?
+  bool IsRunning() const;
 
 private:
-  // Register to handle the signals that indicate when the server should exit.
-  void RegisterSignals();
+  // Register signals which indicate when the server should exit.
+  void AddSignals();
 
-  // Initiate an asynchronous accept operation.
-  void DoAccept();
+  // Wait for a signal to stop the server.
+  void AsyncWaitSignals();
 
-  // Wait for a request to stop the server.
-  void DoAwaitStop();
+  // Listen on the given port.
+  bool Listen(std::uint16_t port);
+
+  // Accept connections asynchronously.
+  void AsyncAccept();
+
+  // Stop acceptor and worker threads, close all pending connections, and
+  // finally stop the event loop.
+  void DoStop();
 
   // Worker thread routine.
   void WorkerRoutine();
@@ -84,6 +103,18 @@ private:
     Strings methods;
   };
 
+  // Port number.
+  std::uint16_t port_;
+
+  // The directory with the static files to be served.
+  Path doc_root_;
+
+  // Is the server running?
+  bool running_;
+
+  // The mutex for guarding the state of the server.
+  std::mutex state_mutex_;
+
   // The io_context used to perform asynchronous operations.
   boost::asio::io_context io_context_;
 
@@ -99,9 +130,6 @@ private:
   // Worker threads.
   std::vector<std::thread> worker_threads_;
 
-  // The directory with the static files to be served.
-  Path doc_root_;
-
   // The queue with connection waiting for the workers to process.
   Queue<ConnectionPtr> queue_;