feat: refactor main event loop and add cgi fds to epoll watchlist

Author: iboukhss

Makefile

@@ -4,7 +4,7 @@ CONFIG ?= debug
 
 # Compiler settings
 CXX = clang++
-CXXFLAGS = -std=c++98 -Wall -Wextra -Wshadow -Werror
+CXXFLAGS = -std=c++98 -Wall -Wextra -Werror
 CPPFLAGS = -Isrc -Ithird_party/utest -MMD -MP
 
 # Build specific options
@@ -37,6 +37,8 @@ srcs = \
   src/core/server_defaults.hpp \
   src/core/signals.cpp \
   src/core/signals.hpp \
+  src/core/virtual_server.cpp \
+  src/core/virtual_server.hpp \
   src/handler/cgi_handler.cpp \
   src/handler/cgi_handler.hpp \
   src/handler/delete_handler.cpp \
@@ -63,15 +65,15 @@ srcs = \
   src/router/router.cpp \
   src/router/router.hpp \
   src/util/itoa.cpp \
-  src/util/to_string.cpp \
-  src/util/to_string.hpp \
   src/util/log_message.cpp \
   src/util/log_message.hpp \
   src/util/str_split.cpp \
   src/util/str_trim.cpp \
   src/util/string.hpp \
   src/util/syscall_error.cpp \
   src/util/syscall_error.hpp \
+  src/util/to_string.cpp \
+  src/util/to_string.hpp \
 
 cpps = $(filter %.cpp,$(srcs))
 hpps = $(filter %.hpp,$(srcs))

src/core/client.cpp

@@ -1,21 +1,333 @@
 #include "core/client.hpp"
 
+#include "core/virtual_server.hpp"
+#include "http/http_parser.hpp"
+#include "util/log_message.hpp"
+
 #include <fcntl.h>
+#include <sys/epoll.h>
 #include <unistd.h>
 
-Client::Client(int fd)
-    : state_(kReceivingHeaders),
-      fd_(fd),
+#include <cerrno>
+#include <cstring>
+#include <iostream>
+
+Client::Client(int fd, const VirtualServer& vs)
+    : sockfd_(fd),
+      vserv_(vs),
+      state_(Client::kReceivingHeaders),
       handler_(NULL)
 {
+    epoll_fds_[sockfd_] = EPOLLIN | EPOLLRDHUP;
+    pipefd_[0] = -1;
+    pipefd_[1] = -1;
 }
 
 Client::~Client()
 {
-    if (fd_ != -1) {
-        close(fd_);
+    if (sockfd_ != -1) {
+        close(sockfd_);
     }
     if (handler_) {
         delete handler_;
     }
 }
+
+size_t Client::sendbuf_available_size() const
+{
+    return WEBSERV_MAX_SENDBUF_SIZE - sendbuf_.size();
+}
+
+void Client::want_read(int fd)
+{
+    epoll_fds_[fd] |= EPOLLIN;
+}
+
+void Client::want_write(int fd)
+{
+    epoll_fds_[fd] |= EPOLLOUT;
+}
+
+void Client::stop_read(int fd)
+{
+    epoll_fds_[fd] &= ~EPOLLIN;
+}
+
+void Client::stop_write(int fd)
+{
+    epoll_fds_[fd] &= ~EPOLLOUT;
+}
+
+void Client::on_epollin(int fd)
+{
+    if (fd == sockfd_) {
+        read_from_socket();
+        maybe_create_request_handler();
+        maybe_write_to_regular_file();
+    }
+    else if (fd == pipefd_[0]) {
+        read_from_pipe();
+    }
+    refresh_interest_list();
+}
+
+void Client::on_epollout(int fd)
+{
+    if (fd == sockfd_) {
+        maybe_read_from_regular_file();
+        write_to_socket();
+    }
+    else if (fd == pipefd_[1]) {
+        write_to_pipe();
+    }
+    refresh_interest_list();
+}
+
+void Client::on_sync_complete()
+{
+    if (state_ == Client::kPreparingNextRequest) {
+        parser_.reset_for_next_request();
+
+        delete handler_;
+        handler_ = NULL;
+
+        epoll_fds_.clear();
+        epoll_fds_[sockfd_] = EPOLLIN | EPOLLRDHUP;
+        pipefd_[0] = -1;
+        pipefd_[1] = -1;
+
+        state_ = Client::kReceivingHeaders;
+    }
+}
+
+// TODO: get rid of this
+static void print_raw_data(const char* s, size_t n)
+{
+    std::cout << "< ";
+
+    for (size_t i = 0; i < n; ++i) {
+        char c = s[i];
+        if (c == '\r') {
+            std::cout << "\\r";
+        }
+        else if (c == '\n') {
+            std::cout << "\\n\n";
+            if (i < n - 1) {
+                std::cout << "< ";
+            }
+        }
+        else {
+            std::cout.put(c);
+        }
+    }
+    std::cout.flush();
+}
+
+void Client::read_from_socket()
+{
+    if (state_ != Client::kReceivingHeaders && state_ != Client::kProcessingRequest)
+        return;
+
+    char buf[4096];
+
+    while (!parser_.has_error() && !parser_.is_done() && parser_.available_size() > 0) {
+        size_t max = std::min(parser_.available_size(), sizeof(buf));
+
+        ssize_t n = recv(sockfd_, buf, max, 0);
+        if (n == -1) {
+            if (errno == EAGAIN || errno == EWOULDBLOCK) {
+                LOG(DEBUG) << "Client #" << sockfd_ << ": recv blocked, retrying";
+                return;
+            }
+            LOG(ERROR) << "Client #" << sockfd_ << ": recv failed: " << strerror(errno);
+            state_ = Client::kClosingConnection;
+            return;
+        }
+        if (n == 0) {
+            LOG(WARN) << "Client #" << sockfd_ << ": Connection closed by remote peer";
+            state_ = Client::kClosingConnection;
+            return;
+        }
+
+        parser_.append_data(buf, n);
+        print_raw_data(buf, n);
+    }
+
+    if (parser_.has_error()) {
+        LOG(ERROR) << "Client #" << sockfd_ << ": Invalid HTTP request";
+        state_ = Client::kClosingConnection;
+        return;
+    }
+}
+
+void Client::write_to_socket()
+{
+    if (state_ != Client::kProcessingRequest)
+        return;
+
+    if (sendbuf_.empty())
+        return;
+
+    ssize_t n = send(sockfd_, sendbuf_.data(), sendbuf_.size(), 0);
+    if (n == -1) {
+        if (errno == EAGAIN || errno == EWOULDBLOCK) {
+            LOG(WARN) << "Client #" << sockfd_ << ": send blocked, retrying";
+            return;
+        }
+        LOG(ERROR) << "Client #" << sockfd_ << ": send failed: " << strerror(errno);
+        state_ = Client::kClosingConnection;
+        return;
+    }
+    if (n == 0) {
+        LOG(WARN) << "Client #" << sockfd_ << "Failed to send any bytes!";
+        return;
+    }
+
+    sendbuf_.erase(0, n);
+
+    if (handler_->is_done() && sendbuf_.empty()) {
+        if (keep_alive()) {
+            state_ = Client::kPreparingNextRequest;
+        }
+        else {
+            state_ = Client::kClosingConnection;
+        }
+    }
+}
+
+static void log_request_info(int fd, const HttpRequest& request)
+{
+    LOG(INFO) << "Client #" << fd << ": " << request.method << " " << request.path << " "
+              << http_version_to_string(request.http_version);
+}
+
+void Client::maybe_create_request_handler()
+{
+    if (state_ != Client::kReceivingHeaders)
+        return;
+
+    if (!parser_.did_parse_headers())
+        return;
+
+    const HttpRequest& req = parser_.request();
+    log_request_info(sockfd_, req);
+
+    handler_ = vserv_.router().handle_request(req);
+    pipefd_[0] = handler_->cgi_read_fd();
+    pipefd_[1] = handler_->cgi_write_fd();
+
+    if (pipefd_[0] != -1) {
+        epoll_fds_[pipefd_[0]] = EPOLLIN;
+    }
+    if (pipefd_[1] != -1) {
+        epoll_fds_[pipefd_[1]] = EPOLLOUT;
+    }
+
+    state_ = Client::kProcessingRequest;
+}
+
+void Client::maybe_write_to_regular_file()
+{
+    if (state_ == Client::kProcessingRequest) {
+        assert(handler_ != NULL);
+
+        if (handler_->is_regular_file()) {
+            write_to_virtual_file();
+        }
+    }
+}
+
+void Client::maybe_read_from_regular_file()
+{
+    if (state_ == Client::kProcessingRequest) {
+        assert(handler_ != NULL);
+
+        if (handler_->is_regular_file()) {
+            read_from_virtual_file();
+        }
+    }
+}
+
+void Client::read_from_pipe()
+{
+    assert(state_ == Client::kProcessingRequest);
+    assert(handler_ != NULL);
+    assert(handler_->cgi_read_fd() != -1);
+
+    read_from_virtual_file();
+}
+
+void Client::write_to_pipe()
+{
+    assert(state_ == Client::kProcessingRequest);
+    assert(handler_ != NULL);
+    assert(handler_->cgi_write_fd() != -1);
+
+    write_to_virtual_file();
+}
+
+void Client::read_from_virtual_file()
+{
+    char buf[4096];
+
+    while (!handler_->is_done() && handler_->has_output() && sendbuf_available_size() > 0) {
+        size_t max = std::min(sizeof(buf), sendbuf_available_size());
+        size_t n = handler_->read_output(buf, max);
+        if (n == 0) {
+            break;
+        }
+        sendbuf_.append(buf, n);
+    }
+}
+
+void Client::write_to_virtual_file()
+{
+    char buf[4096];
+
+    while (!handler_->is_done() && handler_->needs_input() && parser_.has_body_chunk()) {
+        size_t n = parser_.read_next_body_chunk(buf, sizeof(buf));
+        size_t written = handler_->write_input(buf, n);
+        parser_.consume_body_chunk(written);
+        if (written < n) {
+            break;
+        }
+    }
+}
+
+void Client::refresh_interest_list()
+{
+    for (std::map<int, uint32_t>::iterator it = epoll_fds_.begin(); it != epoll_fds_.end(); ++it) {
+        it->second = 0;
+    }
+
+    if (state_ == Client::kClosingConnection) {
+        return;
+    }
+    else if (state_ == Client::kReceivingHeaders) {
+        want_read(sockfd_);
+    }
+    else if (state_ == Client::kProcessingRequest) {
+        assert(handler_ != NULL);
+
+        if (handler_->needs_input()) {
+            want_read(sockfd_);
+        }
+        if (handler_->has_output()) {
+            want_write(sockfd_);
+        }
+        if (pipefd_[0] != -1 && handler_->has_output()) {
+            want_read(pipefd_[0]);
+        }
+        if (pipefd_[1] != -1 && handler_->needs_input()) {
+            want_write(pipefd_[1]);
+        }
+    }
+    else if (state_ == Client::kPreparingNextRequest) {
+        want_read(sockfd_);
+    }
+    else {
+        assert(0 && "UNREACHABLE");
+    }
+
+    epoll_fds_[sockfd_] |= EPOLLRDHUP;
+}