event loop with threadmgr

CHANGELOG.md

@@ -1,4 +1,6 @@
 # 0.3.0 
+- SSL support via OpenSSL
+- Switched from thread per connection to event driven threading model
 - Rewrote request parser for readability and speed
 - Added improved logging with different log levels
 - Separated anthracite into libanthracite and anthracite-bin to allow for other projects to implement anthracite (example in ./src/api_main.cpp)

lib/anthracite.cpp

@@ -1,18 +1,12 @@
 #include "./anthracite.hpp"
 #include "./log/log.hpp"
-#include "./socket/socket.hpp"
-#include "backends/file_backend.hpp"
-#include <chrono>
-#include <condition_variable>
 #include <iostream>
-#include <mutex>
 #include <netinet/in.h>
-#include <span>
 #include <sys/socket.h>
-#include <thread>
 #include <unistd.h>
-#include "./socket/openssl_socket.hpp"
-#include <future>
+#include "./config/config.hpp"
+#include "./thread_mgr/event_loop.hpp"
+#include <signal.h>
 
 using namespace anthracite;
 
@@ -21,103 +15,26 @@ void log_request_and_response(http::request& req, std::unique_ptr<http::response
 constexpr int default_port = 80;
 constexpr int max_worker_threads = 128;
 
-using std::chrono::high_resolution_clock;
-using std::chrono::duration_cast;
-using std::chrono::duration;
-using std::chrono::milliseconds;
+thread_mgr::event_loop* elp = nullptr;
 
-void handle_client(socket::anthracite_socket* s, backends::backend& b, backends::file_backend& fb, std::mutex& thread_wait_mutex, std::condition_variable& thread_wait_condvar, int& active_threads)
-{
-    while (true) {
-        std::string raw_request = s->recv_message(http::HEADER_BYTES);
 
-        // We're doing the start here even though it would ideally be done 
-        // before the first line since if we leave the connection open for 
-        // HTTP 1.1, we can spend a bit of time waiting
-        auto start = high_resolution_clock::now();
-
-        if (raw_request == "") {
-            break;
-        }
-
-        http::request req(raw_request, s->get_client_ip());
-        std::unique_ptr<http::response> resp = req.is_supported_version() ? b.handle_request(req) : fb.handle_error(http::status_codes::HTTP_VERSION_NOT_SUPPORTED);
-        std::string header = resp->header_to_string();
-        s->send_message(header);
-        s->send_message(resp->content());
-
-        auto end = high_resolution_clock::now();
-        auto ms_int = duration_cast<std::chrono::microseconds>(end-start);
-        log_request_and_response(req, resp , ms_int.count());
-        
-        resp.reset();
-        if (req.close_connection()) {
-            break;
-        }
-    }
-    s->close_conn();
-    delete s;
-    {
-        std::lock_guard<std::mutex> lock(thread_wait_mutex);
-        active_threads--;
-    }
-    thread_wait_condvar.notify_one();
+extern "C" void signalHandler(int signum) {
+    log::warn << "Caught signal #" << signum << ", exiting Anthracite" << std::endl;
+    elp->stop();
 }
 
-int listen_loop(int port_number, backends::backend& be, bool tls) {
-    socket::anthracite_socket* socket;
 
-    if (tls){
-        socket = new socket::openssl_socket(port_number);
-    } else {
-        socket = new socket::anthracite_socket(port_number);
-    }
-
-    backends::file_backend fb("./www");
-    log::info << "Listening for " << (tls ? "HTTPS" : "HTTP" ) << " connections on port " << port_number << std::endl;
-     
-    int active_threads = 0;
-    std::mutex thread_wait_mutex;
-    std::condition_variable thread_wait_condvar;
-
-    while (true) {
-        socket->wait_for_conn();
-        std::unique_lock<std::mutex> lock(thread_wait_mutex);
-        thread_wait_condvar.wait(lock, [active_threads] { return active_threads < max_worker_threads; });
-        active_threads++;
-
-        socket::anthracite_socket* client_sock;
-    
-        if (tls){
-            socket::openssl_socket* ssl_sock = dynamic_cast<socket::openssl_socket*>(socket);
-            client_sock = new socket::openssl_socket(*ssl_sock);
-        } else {
-            client_sock = new socket::anthracite_socket(*socket);
-        }
-
-        std::thread(handle_client, client_sock, std::ref(be), std::ref(fb), std::ref(thread_wait_mutex), std::ref(thread_wait_condvar), std::ref(active_threads)).detach();
-    }
-
-    delete socket;
-}
-
-int anthracite_main(int argc, char** argv, backends::backend& be)
+int anthracite_main(backends::backend& be, config::config& config)
 {
+    signal(SIGTERM, signalHandler);
+    signal(SIGSEGV, signalHandler);
+    signal(SIGINT, signalHandler);
+    signal(SIGABRT, signalHandler);
+
     log::logger.initialize(log::LOG_LEVEL_INFO);
-    auto args = std::span(argv, size_t(argc));
-
-    std::vector<std::thread> threads;
-
-    if (argc > 1) {
-        auto thread = std::thread(listen_loop, atoi(argv[1]), std::ref(be), false);
-        thread.detach();
-        threads.push_back(std::move(thread));
-    } if (argc > 2) {
-        auto thread = std::thread(listen_loop, atoi(argv[2]), std::ref(be), true);
-        thread.detach();
-        threads.push_back(std::move(thread));
-    }
-    std::promise<void>().get_future().wait();
+    thread_mgr::event_loop el(be, config);
+    elp = &el;
+    el.start();
     return 0;
 }
 

lib/anthracite.hpp

@@ -1,4 +1,6 @@
 #include "backends/backend.hpp"
+#include "config/config.hpp"
 
 using namespace anthracite;
-int anthracite_main(int argc, char** argv, backends::backend& be);
+
+int anthracite_main(backends::backend& be, config::config& cfg);

lib/socket/openssl_socket.cpp

@@ -45,22 +45,33 @@ openssl_socket::openssl_socket(int port, int max_queue)
 
 openssl_socket::~openssl_socket() = default;
 
-void openssl_socket::wait_for_conn()
+bool openssl_socket::wait_for_conn()
 {
     client_ip = "";
-    client_socket = accept(server_socket, reinterpret_cast<struct sockaddr*>(&client_addr), &client_addr_len);
-    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());
- 
-    _ssl = SSL_new(_context);
-    SSL_set_fd(_ssl, client_socket);
-    if (SSL_accept(_ssl) <= 0) {
-        log::warn << "Unable to open SSL connection with client" << std::endl;
-        client_ip = "";
-        close(client_socket);
-        client_socket = -1;
+    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);
+        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());
+        _ssl = SSL_new(_context);
+        SSL_set_fd(_ssl, client_socket);
+        if (SSL_accept(_ssl) <= 0) {
+            log::warn << "Unable to open SSL connection with client" << std::endl;
+            client_ip = "";
+            close(client_socket);
+            client_socket = -1;
+            return false;
+        }
+        return true;
+    } else {
+        return false;
     }
+
+ 
 }
 
 void openssl_socket::close_conn()

lib/socket/openssl_socket.hpp

@@ -14,7 +14,7 @@ class openssl_socket : public anthracite_socket {
         openssl_socket(int port, int max_queue = MAX_QUEUE_LENGTH);
         ~openssl_socket();
 
-        void wait_for_conn() override;
+        bool wait_for_conn() override;
         void close_conn() override;
         void send_message(std::string& msg) override;
         std::string recv_message(int buffer_size) override;

lib/socket/socket.cpp

@@ -29,13 +29,22 @@ anthracite_socket::anthracite_socket(int port, int max_queue)
     listen(server_socket, max_queue);
 }
 
-void anthracite_socket::wait_for_conn()
+bool anthracite_socket::wait_for_conn()
 {
     client_ip = "";
-    client_socket = accept(server_socket, reinterpret_cast<struct sockaddr*>(&client_addr), &client_addr_len);
-    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());
+    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);
+        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());
+        return true;
+    } else {
+        return false;
+    }
 }
 
 const std::string& anthracite_socket::get_client_ip()

lib/socket/socket.hpp

@@ -14,6 +14,7 @@ namespace anthracite::socket {
 class anthracite_socket {
 
 protected:
+    struct timeval wait_timeout = { .tv_sec = 1, .tv_usec = 0};
     int server_socket;
     int client_socket {};
     std::string client_ip;
@@ -27,7 +28,7 @@ public:
 
     virtual const std::string& get_client_ip() final;
 
-    virtual void wait_for_conn();
+    virtual bool wait_for_conn();
     virtual void close_conn();
     virtual void send_message(std::string& msg);
     virtual std::string recv_message(int buffer_size);

src/api_main.cpp

@@ -108,5 +108,5 @@ int main(int argc, char** argv)
     auto args = std::span(argv, size_t(argc));
     api_backend ab;
     ab.register_endpoint("users/*", handle_request);
-    anthracite_main(argc, argv, ab);
+    //anthracite_main(argc, argv, ab);
 }

src/file_main.cpp

@@ -1,10 +1,15 @@
 #include "../lib/anthracite.hpp"
 #include "../lib/backends/file_backend.hpp"
+#include "../lib/config/config.hpp"
 
 using namespace anthracite;
 
 int main(int argc, char** argv)
 {
     backends::file_backend fb("./www");
-    anthracite_main(argc, argv, fb);
+    config::config cfg(5);
+    cfg.add_http_config(config::http_config(8080));
+    cfg.add_https_config(config::https_config(8081, "", ""));
+
+    anthracite_main(fb, cfg);
 }