Support server workers with EPOLL

Author: EnqueueDuck

README.md

@@ -6,15 +6,33 @@ server to achieve beter through-put.
 
 # Releases
 
-### Version 1.1
-
-To run the server:
+To run the server, default host is 14396:
 
 ```
 bazel run -c opt src/main/cpp:server
 ```
 
-Simple HTTP Server with a multi-threaded backend. The backend employs a ThreadPool, which manages
+### Version 2.0
+
+HTTP Server with N workers. Connections are distributed to the workers, each work handles IO
+using EPOLL.
+
+```
+user~/wrk (master) ./wrk -t8 -c 10000 -d20s http://host:14396
+Running 20s test @ http://host:14396
+  8 threads and 10000 connections
+  Thread Stats   Avg      Stdev     Max   +/- Stdev
+    Latency    51.63us  129.97us  48.81ms   99.68%
+    Req/Sec   103.74k    11.08k  134.55k    71.14%
+  2073920 requests in 20.10s, 79.11MB read
+  Socket errors: connect 8987, read 0, write 0, timeout 0
+Requests/sec: 103189.69
+Transfer/sec:      3.94MB
+```
+
+### Version 1.1
+
+HTTP Server with a multi-threaded backend. The backend employs a ThreadPool, which manages
 N worker threads. Each request is encapsulated in a Task object and passed to one of the worker
 threads for handling.
 

src/main/cpp/BUILD

@@ -34,6 +34,7 @@ cc_binary(
     name = "server",
     srcs = ["main.cpp"],
     linkopts = ["-lpthread"],
+    copts = ["-std=c++1z"],
     deps = [
         ":http",
         "@third_party-glog//:glog",

src/main/cpp/http/http_server.cpp

@@ -4,7 +4,7 @@
 
 namespace http {
 
-void HttpServer::InternalHandle(int connection_sd) {
+void HttpServer::Worker::HandleConnection(int connection_sd) {
   char buffer[opts_.max_http_request_size];
 
   int bytes_read = 1;
@@ -17,9 +17,6 @@ void HttpServer::InternalHandle(int connection_sd) {
     }
     if (bytes_read == 0) break;
 
-    VLOG(1) << "Received " << bytes_read << " bytes";
-    VLOG(1) << buffer;
-
     // Parse the http request
     request::HttpRequest request;
     request.ParseFromString(buffer);
@@ -34,8 +31,66 @@ void HttpServer::InternalHandle(int connection_sd) {
   close(connection_sd);
 }
 
+void HttpServer::Worker::Stop() {
+  stopped_ = true;
+  thread_.join();
+}
+
+void HttpServer::Worker::Initialize() {
+  epoll_fd_ = epoll_create1(0);
+  thread_ = std::thread([this]() {
+    struct epoll_event events[MAX_EPOLL_EVENTS];
+    int num_events;
+    while (!stopped_) {
+      num_events = epoll_wait(epoll_fd_, events, MAX_EPOLL_EVENTS, EPOLL_TIMEOUT_MS);
+      for (int i = 0; i < num_events; ++i) {
+        // Client hang up
+        if (events[i].events & EPOLLHUP) {
+          close(events[i].data.fd);
+          continue;
+        }
+
+        // Error
+        if ((events[i].events & EPOLLERR) || !(events[i].events & EPOLLIN)) {
+          LOG(ERROR) << "Error caught with fd " << events[i].data.fd;
+          continue;
+        }
+
+        HandleConnection(events[i].data.fd);
+      }
+    }
+
+  });
+}
+
+void HttpServer::Worker::AddConnection(int connection_sd) {
+  struct epoll_event event;
+  event.data.fd = connection_sd;
+  event.events = EPOLLIN;
+
+  epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, connection_sd, &event);
+}
+
 void HttpServer::Handle(int connection_sd) {
-  executor_.Execute(std::bind(&HttpServer::InternalHandle, this, connection_sd));
+  auto &worker = workers_[current_worker_idx_++];
+  worker->AddConnection(connection_sd);
+  if (current_worker_idx_ >= workers_.size()) current_worker_idx_ = 0;
+}
+
+void HttpServer::Initialize() {
+  Server::Initialize();
+  for (auto &worker : workers_) {
+    worker->Initialize();
+  }
+}
+
+response::HttpResponse HttpServer::Worker::HandleRequest(
+    const request::HttpRequest &request
+) const {
+  response::HttpResponse response;
+  response.status_code = response::HTTP_200_OK;
+  response.body = "ok";
+  return response;
 }
 
 } // namespace http

src/main/cpp/http/http_server.h

@@ -1,46 +1,74 @@
 #pragma once
 
+#include <sys/epoll.h>
+#include <thread>
+#include <atomic>
+#include <memory>
+
 #include "http/server.h"
-#include "thread/executor.h"
 #include "http/http_response.h"
 #include "http/http_request.h"
 
 namespace http {
 
 struct HttpServerOpts {
-  constexpr static int DEFAULT_MAX_HTTP_REQUEST_SIZE = 65536;
-
   ServerOpts server_opts;
-  thread::ExecutorOpts executor_opts;
-
-  int max_http_request_size = DEFAULT_MAX_HTTP_REQUEST_SIZE;
+  int num_workers;
 };
 
 class HttpServer : public Server {
  public:
-  explicit HttpServer(HttpServerOpts opts)
-      : Server(opts.server_opts), opts_(opts), executor_(opts.executor_opts) { }
+  constexpr static int DEFAULT_HTTP_REQUEST_SIZE = 1024;
 
-  void Initialize() {
-    Server::Initialize();
-    executor_.Initialize();
+  explicit HttpServer(const HttpServerOpts &opts) : Server(opts.server_opts), opts_(opts) {
+    for (int i = 0; i < opts_.num_workers; ++i) {
+      workers_.emplace_back(std::make_unique<Worker>(WorkerOpts()));
+    }
+    LOG(INFO) << "Started " << opts_.num_workers << " workers";
   }
 
+  struct WorkerOpts {
+    int max_http_request_size = DEFAULT_HTTP_REQUEST_SIZE;
+  };
+
+  class Worker {
+   public:
+    Worker() = default;
+    explicit Worker(const WorkerOpts &opts) : opts_(opts) { }
+
+    constexpr static int MAX_EPOLL_EVENTS = 64;
+    constexpr static int EPOLL_TIMEOUT_MS = 10000; // 10 seconds
+
+    void Initialize();
+
+    void Stop();
+
+    void AddConnection(int connection_sd);
+
+   private:
+    WorkerOpts opts_;
+
+    int epoll_fd_;
+    std::thread thread_;
+    std::atomic_bool stopped_{false};
+
+    void HandleConnection(int connection_sd);
+
+    virtual response::HttpResponse HandleRequest(const request::HttpRequest &request) const;
+  };
+
+  using WorkerPtr = std::unique_ptr<Worker>;
+
+  void Initialize();
+
  protected:
   void Handle(int connection_sd) override;
 
-  virtual response::HttpResponse HandleRequest(const request::HttpRequest &request) const {
-    response::HttpResponse response;
-    response.status_code = response::HTTP_200_OK;
-    response.body = "ok";
-    return response;
-  }
 
  private:
   HttpServerOpts opts_;
-  thread::Executor executor_;
-
-  void InternalHandle(int connection_sd);
+  std::vector<WorkerPtr> workers_;
+  int current_worker_idx_ = 0;
 };
 
 } // namespace http

src/main/cpp/main.cpp

@@ -11,7 +11,7 @@ int main(int argc, char *argv[]) {
   google::ParseCommandLineFlags(&argc, &argv, true);
 
   http::HttpServerOpts opts;
-  opts.executor_opts.num_worker_threads = std::thread::hardware_concurrency();
+  opts.num_workers = std::thread::hardware_concurrency();
 
   http::HttpServer server(opts);
   server.Initialize();

src/main/cpp/tests/test_server.cpp

@@ -11,7 +11,7 @@ int main(int argc, char *argv[]) {
   google::ParseCommandLineFlags(&argc, &argv, true);
 
   http::HttpServerOpts opts;
-  opts.executor_opts.num_worker_threads = std::thread::hardware_concurrency();
+  opts.num_workers = std::thread::hardware_concurrency();
 
   http::HttpServer server(opts);
   server.Initialize();