faster nonblocking io

Dockerfile

@@ -1,8 +1,9 @@
 FROM alpine as build-env
 
-RUN apk add --no-cache build-base python3 cmake
+RUN apk add --no-cache build-base python3 cmake openssl-dev
 COPY ./src ./src
 COPY ./lib ./lib
+COPY ./tests ./tests
 COPY ./default_www ./default_www
 COPY ./build_supp ./build_supp
 COPY ./CMakeLists.txt .

docker-compose.yml

@@ -2,7 +2,7 @@ services:
   anthracite-web:
     build: .
     ports:
-      - "8080:80"
+      - "8080:8080"
     volumes:
       - type: bind
         source: ./default_www/docker_compose/

lib/log/log.cpp

@@ -1,4 +1,5 @@
 #include "./log.hpp"
+#include <syncstream>
 
 namespace anthracite::log {
 enum LOG_LEVEL Logger::_level = LOG_LEVEL_NONE;
@@ -25,7 +26,9 @@ LogBuf::LogBuf(std::ostream& output_stream, const std::string& tag, enum LOG_LEV
 int LogBuf::sync()
 {
     if (this->_level <= logger._level) {
-        std::cout << "[" << this->_tag << "] " << this->str();
+        char thread_name[100];
+        pthread_getname_np(pthread_self(), thread_name, 100);
+        std::osyncstream(std::cout) << "[" << this->_tag << "] [" << syscall(SYS_gettid) <<  ":" << thread_name << "] "<< this->str();
         std::cout.flush();
     }
     this->str("");

lib/socket/socket.cpp

@@ -4,18 +4,26 @@
 #include <malloc.h>
 #include <netinet/in.h>
 #include <string>
+#include <sys/epoll.h>
 #include <sys/socket.h>
 #include <sys/time.h>
 #include <unistd.h>
 #include <vector>
+#include <unistd.h>
+#include <fcntl.h>
+#include "assert.h"
+#include <netinet/in.h>
+#include <netinet/tcp.h>
+
 
 namespace anthracite::socket {
 
 const struct timeval anthracite_socket::timeout_tv = { .tv_sec = 5, .tv_usec = 0 };
 
-anthracite_socket::anthracite_socket(int port, int max_queue)
+anthracite_socket::anthracite_socket(int port, int max_queue, bool nonblocking)
     : server_socket(::socket(AF_INET, SOCK_STREAM, 0))
     , client_ip("")
+    , _nonblocking(nonblocking)
 {
     struct sockaddr_in address {};
     address.sin_family = AF_INET;
@@ -26,18 +34,21 @@ anthracite_socket::anthracite_socket(int port, int max_queue)
     setsockopt(server_socket, SOL_SOCKET, SO_REUSEADDR, &reuse_opt, sizeof(reuse_opt));
     bind(server_socket, reinterpret_cast<struct sockaddr*>(&address), sizeof(address));
 
+    if (_nonblocking) {
+        fcntl(server_socket, F_SETFL, O_NONBLOCK);
+    }
+
     listen(server_socket, max_queue);
 }
 
 bool anthracite_socket::wait_for_conn()
 {
     client_ip = "";
-    struct timeval tv = { .tv_sec = 1, .tv_usec = 0 };
-    fd_set read_fd;
-    FD_ZERO(&read_fd);
-    FD_SET(server_socket, &read_fd);
-    if (select(server_socket + 1, &read_fd, NULL, NULL, &wait_timeout)) {
-        client_socket = accept(server_socket, reinterpret_cast<struct sockaddr*>(&client_addr), &client_addr_len);
+    client_socket = accept(server_socket, reinterpret_cast<struct sockaddr*>(&client_addr), &client_addr_len);
+    if (client_socket > 0) {
+        if (_nonblocking) {
+            fcntl(client_socket, F_SETFL, O_NONBLOCK);
+        }
         std::array<char, INET_ADDRSTRLEN> ip_str { 0 };
         inet_ntop(AF_INET, &client_addr.sin_addr, ip_str.data(), INET_ADDRSTRLEN);
         client_ip = std::string(ip_str.data());
@@ -52,8 +63,7 @@ const std::string& anthracite_socket::get_client_ip()
     return client_ip;
 }
 
-void anthracite_socket::close_conn()
-{
+void anthracite_socket::close_conn() {
     close(client_socket);
     client_socket = -1;
 }
@@ -66,13 +76,21 @@ void anthracite_socket::send_message(std::string& msg)
     send(client_socket, &msg[0], msg.length(), 0);
 }
 
+bool anthracite_socket::has_client() {
+    return client_socket > 0;
+}
+
 std::string anthracite_socket::recv_message(int buffer_size)
 {
     if (client_socket == -1) {
         return "";
     }
 
-    setsockopt(client_socket, SOL_SOCKET, SO_RCVTIMEO, &timeout_tv, sizeof timeout_tv);
+    //setsockopt(client_socket, SOL_SOCKET, SO_RCVTIMEO, &timeout_tv, sizeof timeout_tv);
+
+    int nodelay_opt = 1;
+    assert(setsockopt(client_socket, SOL_TCP, TCP_NODELAY, &nodelay_opt, sizeof(nodelay_opt)) == 0);
+
     std::vector<char> response(buffer_size + 1);
     ssize_t result = recv(client_socket, response.data(), buffer_size + 1, 0);
 

lib/socket/socket.hpp

@@ -14,6 +14,7 @@ namespace anthracite::socket {
 class anthracite_socket {
 
 protected:
+    bool _nonblocking;
     struct timeval wait_timeout = { .tv_sec = 1, .tv_usec = 0};
     int server_socket;
     int client_socket {};
@@ -24,10 +25,11 @@ protected:
     static const int MAX_QUEUE_LENGTH = 100;
 
 public:
-    anthracite_socket(int port, int max_queue = MAX_QUEUE_LENGTH);
+    anthracite_socket(int port, int max_queue = MAX_QUEUE_LENGTH, bool nonblocking = false);
 
     virtual const std::string& get_client_ip() final;
 
+    virtual bool has_client();
     virtual bool wait_for_conn();
     virtual void close_conn();
     virtual void send_message(std::string& msg);

lib/thread_mgr/event_loop.cpp

@@ -5,7 +5,10 @@
 #include "sys/epoll.h"
 #include <chrono>
 #include <mutex>
+#include <pthread.h>
+#include <sstream>
 #include <syncstream>
+#include "signal.h"
 
 using std::chrono::duration;
 using std::chrono::duration_cast;
@@ -35,25 +38,20 @@ event_loop::event_loop(backends::backend& backend, config::config& config)
 {
 }
 
-bool event_loop::event_handler(event& event)
+bool event_loop::event_handler(socket::anthracite_socket* sock)
 {
-    std::string raw_request = event.socket()->recv_message(http::HEADER_BYTES);
+    std::string raw_request = sock->recv_message(http::HEADER_BYTES);
 
     if (raw_request == "") {
         return false;
     }
 
-    http::request req(raw_request, event.socket()->get_client_ip());
+    http::request req(raw_request, sock->get_client_ip());
     std::unique_ptr<http::response> resp = req.is_supported_version() ? _backend.handle_request(req) : _error_backend.handle_error(http::status_codes::HTTP_VERSION_NOT_SUPPORTED);
     std::string header = resp->header_to_string();
-    event.socket()->send_message(header);
-    event.socket()->send_message(resp->content());
+    sock->send_message(header);
+    sock->send_message(resp->content());
 
-    auto end = high_resolution_clock::now();
-    auto ms_int = duration_cast<std::chrono::microseconds>(end - event.timestamp());
-    log::logger.log_request_and_response(req, resp, ms_int.count());
-
-    resp.reset();
     if (req.close_connection()) {
         return false;
     }
@@ -63,70 +61,36 @@ bool event_loop::event_handler(event& event)
 
 void event_loop::worker_thread_loop(int threadno)
 {
-    unsigned char buf[sizeof(class event)];
+    std::stringstream ss;
+    ss << "worker " << threadno;
+    pthread_setname_np(pthread_self(), ss.str().c_str());
 
-    std::osyncstream(log::info) << "Starting worker thread " << threadno << std::endl;
-    while (_run) {
-        // Get event from queue
-        std::unique_lock lock(_event_mtx);
+    struct epoll_event* events = new struct epoll_event[_config.max_clients()];
+    int timeout_ms = 1000;
 
-        event* ev = nullptr;
-
-        if (_events.size() > 0) {
-            if (!lock.owns_lock()) {
-                lock.lock();
-            }
-            assert(lock.owns_lock());
-            ev = new (buf) event(_events.front());
-            _events.pop();
-            lock.unlock();
-        } else {
-            _event_cv.wait(lock, [this] { return this->_events.size() > 0 || !_run; });
-
-            if (!_run) {
-                break;
-            }
-
-            assert(lock.owns_lock());
-            ev = new (buf) event(_events.front());
-            _events.pop();
-            lock.unlock();
-        }
-
-        if (event_handler(*ev)) {
-            struct epoll_event event;
-            event.events = EPOLLIN;
-            event.data.ptr = ev->socket();
-            epoll_ctl(_epoll_fd, EPOLL_CTL_ADD, ev->socket()->csock(), &event);
-        } else {
-            ev->socket()->close_conn();
-            delete ev->socket();
-        }
+    if (_nonblocking) {
+        timeout_ms = 0;
     }
 
-    std::osyncstream(log::info) << "Stopping worker thread " << threadno << std::endl;
-}
+    log::info << "Starting worker thread " << threadno << std::endl;
 
-void event_loop::eventer_thread_loop()
-{
-    struct epoll_event* events = new struct epoll_event[_config.max_clients()];
-    std::osyncstream(log::info) << "epoll() thread started" << std::endl;
     while (_run) {
-        int ready_fds = epoll_wait(_epoll_fd, events, _config.max_clients(), 1000);
+        int ready_fds = epoll_wait(_epoll_fd, events, _config.max_clients(), timeout_ms);
 
         if (ready_fds > 0) {
-            std::lock_guard<std::mutex> lg(_event_mtx);
             for (int i = 0; i < ready_fds; i++) {
                 socket::anthracite_socket* sockptr = reinterpret_cast<socket::anthracite_socket*>(events[i].data.ptr);
-                struct epoll_event ev;
-                epoll_ctl(_epoll_fd, EPOLL_CTL_DEL, sockptr->csock(), &events[i]);
-                _events.push(event(sockptr, std::chrono::high_resolution_clock::now()));
+
+                if (!event_handler(sockptr)) {
+                    epoll_ctl(_epoll_fd, EPOLL_CTL_DEL, sockptr->csock(), &events[i]);
+                    sockptr->close_conn();
+                    delete sockptr;
+                }
             }
-            _event_cv.notify_one();
         }
     }
-    delete[] events;
-    std::osyncstream(log::info) << "epoll() thread exited" << std::endl;
+
+    log::info << "Stopping worker thread " << threadno << std::endl;
 }
 
 void event_loop::listener_thread_loop(config::http_config& http_config)
@@ -158,7 +122,7 @@ void event_loop::listener_thread_loop(config::http_config& http_config)
             }
 
             struct epoll_event event;
-            event.events = EPOLLIN;
+            event.events = EPOLLIN | EPOLLEXCLUSIVE;// | EPOLLET;
             event.data.ptr = client_sock;
             epoll_ctl(_epoll_fd, EPOLL_CTL_ADD, client_sock->csock(), &event);
         }
@@ -171,10 +135,11 @@ void event_loop::listener_thread_loop(config::http_config& http_config)
 
 void event_loop::start()
 {
+    signal(SIGPIPE, SIG_IGN);
     log::info << "Starting event_loop Thread Manager" << std::endl;
 
-    _epoll_fd = epoll_create(1);
     _run = true;
+    _epoll_fd = epoll_create(1);
 
     std::vector<std::thread> listener_threads;
     std::vector<std::thread> worker_threads;
@@ -194,11 +159,6 @@ void event_loop::start()
         listener_threads.push_back(std::move(thread));
     }
 
-    {
-        auto thread = std::thread(&event_loop::eventer_thread_loop, this);
-        listener_threads.push_back(std::move(thread));
-    }
-
     for (std::thread& t : worker_threads) {
         t.join();
     }
@@ -211,7 +171,5 @@ void event_loop::start()
 void event_loop::stop()
 {
     _run = false;
-    std::lock_guard<std::mutex> lg(_event_mtx);
-    _event_cv.notify_all();
 }
 }

lib/thread_mgr/event_loop.hpp

@@ -19,14 +19,13 @@ namespace anthracite::thread_mgr {
 
             int _epoll_fd;
             std::mutex _event_mtx;
-            std::condition_variable _event_cv;
-            std::queue<event> _events;
             backends::file_backend _error_backend;
+            bool _nonblocking;
 
             void worker_thread_loop(int threadno);
             void listener_thread_loop(config::http_config& http_config);
             void eventer_thread_loop();
-            bool event_handler(event& ev);
+            bool event_handler(socket::anthracite_socket*);
 
         public:
             event_loop(backends::backend& backend, config::config& config);

src/file_main.cpp

@@ -10,7 +10,7 @@ std::shared_ptr<anthracite::thread_mgr::thread_mgr> server = nullptr;
 
 extern "C" void signalHandler(int signum)
 {
-    anthracite::log::warn << "Caught signal SIG" << sigabbrev_np(signum) << ", exiting Anthracite" << std::endl;
+    //anthracite::log::warn << "Caught signal SIG" << sigabbrev_np(signum) << ", exiting Anthracite" << std::endl;
     if (server != nullptr) {
         server->stop();
     }
@@ -23,7 +23,7 @@ int main(int argc, char** argv)
     signal(SIGINT, signalHandler);
 
     anthracite::backends::file_backend fb("./www");
-    anthracite::config::config cfg(5, 10000);
+    anthracite::config::config cfg(1, 10);
     cfg.add_http_config(anthracite::config::http_config(8080));
     // cfg.add_https_config(config::https_config(8081, "", ""));