Switch to single-threaded epoll-based operation

buffer.cc

@@ -23,6 +23,14 @@ bool ConstBuffer::Discard(size_t len) {
 	return true;
 }
 
+void ConstBuffer::ResetRead() {
+	start_ = commit_;
+}
+
+void ConstBuffer::Commit() {
+	commit_ = start_;
+}
+
 
 Buffer::Buffer(char *buf, size_t size, size_t len)
 		: ConstBuffer(buf, size),
@@ -58,11 +66,12 @@ void Buffer::Wrote(size_t len) {
 	len_ += len;
 }
 
-void Buffer::Commit() {
+void Buffer::Consume() {
-	if (start_ == 0) {
+	if (commit_ == 0) {
 		return;
 	}
-	memmove(buf_, &buf_[start_], len_ - start_);
+	memmove(buf_, &buf_[commit_], len_ - commit_);
-	len_ -= start_;
+	len_ -= commit_;
-	start_ = 0;
+	start_ -= commit_;
+	commit_ = 0;
 }

buffer.h

@@ -12,11 +12,14 @@ class ConstBuffer {
 	template<class T> [[nodiscard]] const T *ReadObj();
 
 	bool Discard(size_t len); // like Read() but don't use the result
+	void ResetRead(); // next read from last commit
+	void Commit(); // commit read position
 
   protected:
 	const char *const_buf_;
 	size_t len_;
 	size_t start_ = 0;
+	size_t commit_ = 0;
 };
 
 class Buffer : public ConstBuffer {
@@ -29,7 +32,7 @@ class Buffer : public ConstBuffer {
 	bool Write(const std::string_view& str);
 	void Wrote(size_t len);
 
-	void Commit(); // commit read position
+	void Consume(); // discard up to last commit
 
   protected:
   	std::unique_ptr<char> own_buf_;

fastcgi.cc

@@ -1,8 +1,8 @@
 #include <arpa/inet.h>
 #include <netinet/in.h>
 #include <signal.h>
+#include <sys/epoll.h>
 #include <sys/socket.h>
-#include <thread>
 
 #include "fastcgi.h"
 #include "fastcgi_conn.h"
@@ -14,6 +14,9 @@ FastCGIServer::FastCGIServer(int port, const std::function<void(std::unique_ptr<
 
 	signal(SIGPIPE, SIG_IGN);
 
+	epoll_fd_ = epoll_create1(0);
+	PCHECK(epoll_fd_ >= 0) << "epoll_create()";
+
 	listen_sock_ = socket(AF_INET6, SOCK_STREAM, 0);
 	PCHECK(listen_sock_ >= 0) << "socket()";
 
@@ -32,20 +35,58 @@ FastCGIServer::FastCGIServer(int port, const std::function<void(std::unique_ptr<
 	}
 
 	PCHECK(listen(listen_sock_, 128) == 0);
+
+	{
+		struct epoll_event ev{
+			.events = EPOLLIN,
+			.data = {
+				.ptr = nullptr,
+			},
+		};
+		PCHECK(epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, listen_sock_, &ev) == 0);
+	}
 }
 
 void FastCGIServer::Serve() {
 	while (true) {
-		sockaddr_in6 client_addr;
+		struct epoll_event events[256];
-		socklen_t client_addr_len = sizeof(client_addr);
+		auto num_fd = epoll_wait(epoll_fd_, events, sizeof(events), -1);
+		if (num_fd == -1 && errno == EINTR) {
+			continue;
+		}
+		PCHECK(num_fd > 0) << "epoll_wait()";
 
-		auto client_sock = accept(listen_sock_, (sockaddr*) &client_addr, &client_addr_len);
+		for (auto i = 0; i < num_fd; ++i) {
-		PCHECK(client_sock >= 0) << "accept()";
+			if (events[i].data.ptr == nullptr) {
-		CHECK_EQ(client_addr.sin6_family, AF_INET6);
+				NewConn();
+			} else {
+				auto conn = static_cast<FastCGIConn*>(events[i].data.ptr);
+				if (!conn->Read()) {
+					PCHECK(epoll_ctl(epoll_fd_, EPOLL_CTL_DEL, conn->Sock(), nullptr) == 0);
+					delete conn;
+				}
+			}
+		}
+	}
+}
 
+void FastCGIServer::NewConn() {
+	sockaddr_in6 client_addr;
+	socklen_t client_addr_len = sizeof(client_addr);
+
+	auto client_sock = accept(listen_sock_, (sockaddr*) &client_addr, &client_addr_len);
+	PCHECK(client_sock >= 0) << "accept()";
+	CHECK_EQ(client_addr.sin6_family, AF_INET6);
+
+	{
 		auto *conn = new FastCGIConn(client_sock, client_addr, callback_, headers_);
-		std::thread thread([conn]() { conn->Serve(); });
+		struct epoll_event ev{
-		thread.detach();
+			.events = EPOLLIN,
+			.data = {
+				.ptr = conn,
+			},
+		};
+		PCHECK(epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, client_sock, &ev) == 0);
 	}
 }
 

fastcgi.h

@@ -12,7 +12,10 @@ class FastCGIServer {
 	void Serve();
 
   private:
-	int listen_sock_;
+	void NewConn();
+
 	const std::function<void(std::unique_ptr<FastCGIRequest>)> callback_;
 	const std::unordered_set<std::string_view> headers_;
+	int epoll_fd_;
+	int listen_sock_;
 };

fastcgi_conn.cc

@@ -31,16 +31,22 @@ bool FastCGIConn::Write(const std::vector<iovec>& vecs) {
 	return writev(sock_, vecs.data(), vecs.size()) == total_size;
 }
 
-void FastCGIConn::Serve() {
+bool FastCGIConn::Read() {
+	if (!buf_.Refill()) {
+		return false;
+	}
+
 	while (true) {
+		buf_.ResetRead();
+
 		const auto *header = buf_.ReadObj<FastCGIHeader>();
 		if (!header) {
 			break;
 		}
 
 		CHECK_EQ(header->version, 1);
-		if (!buf_.BufferAtLeast(header->ContentLength())) {
+		if (buf_.ReadMaxLen() < header->ContentLength()) {
-			return;
+			break;
 		}
 
 		switch (header->type) {
@@ -95,5 +101,10 @@ void FastCGIConn::Serve() {
 		buf_.Commit(); // we've acted on the bytes read so far
 	}
 
-	delete this;
+	buf_.Consume();
+	return true;
+}
+
+int FastCGIConn::Sock() {
+	return sock_;
 }

fastcgi_conn.h

@@ -14,10 +14,11 @@ class FastCGIConn {
 	FastCGIConn(int sock, const sockaddr_in6& client_addr, const std::function<void(std::unique_ptr<FastCGIRequest>)>& callback, const std::unordered_set<std::string_view>& headers);
 	~FastCGIConn();
 
-	void Serve();
+	[[nodiscard]] bool Read();
-
 	[[nodiscard]] bool Write(const std::vector<iovec>& vecs);
 
+	[[nodiscard]] int Sock();
+
   private:
   	const int sock_;
 	const std::function<void(std::unique_ptr<FastCGIRequest>)>& callback_;

stream_buffer.cc

@@ -4,23 +4,11 @@ StreamBuffer::StreamBuffer(int sock, size_t size)
 		: Buffer(size),
 		  sock_(sock) {}
 
-bool StreamBuffer::BufferAtLeast(size_t len) {
+bool StreamBuffer::Refill() {
-	CHECK_LE(start_ + len, size_);
+	auto read_len = read(sock_, WritePtr(), WriteMaxLen());
-
+	if (read_len == 0) {
-	while (ReadMaxLen() < len) {
+		return false;
-		auto read_len = read(sock_, WritePtr(), WriteMaxLen());
-		PCHECK(read_len >= 0);
-		if (read_len == 0) {
-			return false;
-		}
-		Wrote(read_len);
 	}
+	Wrote(read_len);
 	return true;
 }
-
-const char *StreamBuffer::Read(size_t len) {
-	if (!BufferAtLeast(len)) {
-		return nullptr;
-	}
-	return Buffer::Read(len);
-}

stream_buffer.h

@@ -6,14 +6,8 @@ class StreamBuffer : public Buffer {
   public:
 	StreamBuffer(int sock, size_t size);
 
-	[[nodiscard]] bool BufferAtLeast(size_t len);
+	[[nodiscard]] bool Refill();
-	[[nodiscard]] const char *Read(size_t len) override;
-	template<class T> [[nodiscard]] const T *ReadObj();
 
   private:
   	int sock_;
 };
-
-template<class T> const T *StreamBuffer::ReadObj() {
-	return reinterpret_cast<const T*>(Read(sizeof(T)));
-}