Merge branch 'master' of github.com:flamingcowtv/mirall

example_simple.cc

@@ -4,6 +4,7 @@
 #include "fastcgi.h"
 
 DEFINE_int32(port, 9000, "TCP port to bind");
+DEFINE_int32(threads, 1, "Number of server threads");
 
 int main(int argc, char *argv[]) {
 	google::InitGoogleLogging(argv[0]);
@@ -13,6 +14,6 @@ int main(int argc, char *argv[]) {
 		request->WriteHeader("Content-Type", "text/plain");
 		request->WriteBody("Hello world");
 		request->End();
-	});
+	}, FLAGS_threads);
 	server.Serve();
 }

fastcgi.cc

@@ -3,38 +3,34 @@
 #include <signal.h>
 #include <sys/epoll.h>
 #include <sys/socket.h>
+#include <thread>
 
 #include "fastcgi.h"
 #include "fastcgi_conn.h"
 
-FastCGIServer::FastCGIServer(int port, const std::function<void(std::unique_ptr<FastCGIRequest>)>& callback, const std::unordered_set<std::string_view>& headers)
+FastCGIServer::FastCGIServer(int port, const std::function<void(std::unique_ptr<FastCGIRequest>)>& callback, int threads, const std::unordered_set<std::string_view>& headers)
-		: callback_(callback),
+		: port_(port),
+		  callback_(callback),
+		  threads_(threads),
 		  headers_(headers) {
-	LOG(INFO) << "listening on [::1]:" << port;
+	LOG(INFO) << "listening on [::1]:" << port_;
 
 	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()";
-
-	{
-		int optval = 1;
-		PCHECK(setsockopt(listen_sock_, SOL_SOCKET, SO_REUSEPORT, &optval, sizeof(optval)) == 0);
 }
 
-	{
+void FastCGIServer::Serve() {
-		sockaddr_in6 bind_addr = {
+	std::vector<std::thread> threads;
-			.sin6_family = AF_INET6,
+	for (int i = 0; i < threads_ - 1; ++i) {
-			.sin6_port = htons(port),
+		threads.emplace_back([this]() { ServeInt(); });
-			.sin6_addr = IN6ADDR_LOOPBACK_INIT,
+	}
-		};
+	ServeInt();
-		PCHECK(bind(listen_sock_, (sockaddr*) &bind_addr, sizeof(bind_addr)) == 0);
 }
 
-	PCHECK(listen(listen_sock_, 128) == 0);
+void FastCGIServer::ServeInt() {
+	auto epoll_fd = epoll_create1(0);
+	PCHECK(epoll_fd >= 0) << "epoll_create()";
+
+	auto listen_sock = NewListenSock();
 
 	{
 		struct epoll_event ev{
@@ -43,15 +39,13 @@ FastCGIServer::FastCGIServer(int port, const std::function<void(std::unique_ptr<
 				.ptr = nullptr,
 			},
 		};
-		PCHECK(epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, listen_sock_, &ev) == 0);
+		PCHECK(epoll_ctl(epoll_fd, EPOLL_CTL_ADD, listen_sock, &ev) == 0);
-	}
 	}
 
-void FastCGIServer::Serve() {
 	while (true) {
 		constexpr auto max_events = 256;
 		struct epoll_event events[max_events];
-		auto num_fd = epoll_wait(epoll_fd_, events, max_events, -1);
+		auto num_fd = epoll_wait(epoll_fd, events, max_events, -1);
 		if (num_fd == -1 && errno == EINTR) {
 			continue;
 		}
@@ -59,12 +53,12 @@ void FastCGIServer::Serve() {
 
 		for (auto i = 0; i < num_fd; ++i) {
 			if (events[i].data.ptr == nullptr) {
-				NewConn();
+				NewConn(listen_sock, epoll_fd);
 			} else {
 				auto conn = static_cast<FastCGIConn*>(events[i].data.ptr);
 				auto fd = conn->Read();
 				if (fd != -1) {
-					PCHECK(epoll_ctl(epoll_fd_, EPOLL_CTL_DEL, fd, nullptr) == 0);
+					PCHECK(epoll_ctl(epoll_fd, EPOLL_CTL_DEL, fd, nullptr) == 0);
 					delete conn;
 				}
 			}
@@ -72,11 +66,11 @@ void FastCGIServer::Serve() {
 	}
 }
 
-void FastCGIServer::NewConn() {
+void FastCGIServer::NewConn(int listen_sock, int epoll_fd) {
 	sockaddr_in6 client_addr;
 	socklen_t client_addr_len = sizeof(client_addr);
 
-	auto client_sock = accept(listen_sock_, (sockaddr*) &client_addr, &client_addr_len);
+	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);
 
@@ -88,8 +82,28 @@ void FastCGIServer::NewConn() {
 				.ptr = conn,
 			},
 		};
-		PCHECK(epoll_ctl(epoll_fd_, EPOLL_CTL_ADD, client_sock, &ev) == 0);
+		PCHECK(epoll_ctl(epoll_fd, EPOLL_CTL_ADD, client_sock, &ev) == 0);
 	}
 }
 
+int FastCGIServer::NewListenSock() {
+	auto sock = socket(AF_INET6, SOCK_STREAM, 0);
+	PCHECK(sock >= 0) << "socket()";
 
+	{
+		int optval = 1;
+		PCHECK(setsockopt(sock, SOL_SOCKET, SO_REUSEPORT, &optval, sizeof(optval)) == 0);
+	}
+
+	{
+		sockaddr_in6 bind_addr = {
+			.sin6_family = AF_INET6,
+			.sin6_port = htons(port_),
+			.sin6_addr = IN6ADDR_LOOPBACK_INIT,
+		};
+		PCHECK(bind(sock, (sockaddr*) &bind_addr, sizeof(bind_addr)) == 0);
+	}
+
+	PCHECK(listen(sock, 128) == 0);
+	return sock;
+}

fastcgi.h

@@ -8,14 +8,16 @@
 
 class FastCGIServer {
   public:
-	FastCGIServer(int port, const std::function<void(std::unique_ptr<FastCGIRequest>)>& callback, const std::unordered_set<std::string_view>& headers={});
+	FastCGIServer(int port, const std::function<void(std::unique_ptr<FastCGIRequest>)>& callback, int threads=1, const std::unordered_set<std::string_view>& headers={});
 	void Serve();
 
   private:
-	void NewConn();
+	void NewConn(int listen_sock, int epoll_fd);
+	int NewListenSock();
+	void ServeInt();
 
+	const int port_;
 	const std::function<void(std::unique_ptr<FastCGIRequest>)> callback_;
+	const int threads_;
 	const std::unordered_set<std::string_view> headers_;
-	int epoll_fd_;
-	int listen_sock_;
 };

sse.cc

@@ -4,6 +4,7 @@ SSEServer::SSEServer(int port, const std::function<void(std::unique_ptr<SSEStrea
 		: callback_(callback),
 		  fastcgi_server_(port,
 		                  [this](std::unique_ptr<FastCGIRequest> request) { OnRequest(std::move(request)); },
+						  1,
 						  {"HTTP_ACCEPT"}) {}
 
 void SSEServer::Serve() {