Remove the nested read loops in FastCGIConn, hide complexity inside StreamBuffer

.gitignore

@@ -1,2 +1,3 @@
 mirall
+example_simple
 *.o

Makefile

@@ -1,5 +1,5 @@
-example_simple: example_simple.o fastcgi.o fastcgi_conn.o fastcgi_request.o buffer.o
-	clang++ -std=gnu++2a -o example_simple example_simple.o fastcgi.o fastcgi_conn.o fastcgi_request.o buffer.o -lgflags -lglog -lpthread
+example_simple: example_simple.o fastcgi.o fastcgi_conn.o fastcgi_request.o stream_buffer.o buffer.o
+	clang++ -std=gnu++2a -o example_simple example_simple.o fastcgi.o fastcgi_conn.o fastcgi_request.o stream_buffer.o buffer.o -lgflags -lglog -lpthread
 
 clean:
 	rm --force exmaple_simple *.o

buffer.cpp

@@ -23,14 +23,6 @@ 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),
@@ -57,12 +49,11 @@ void Buffer::Wrote(size_t len) {
 	len_ += len;
 }
 
-void Buffer::Consume() {
-	if (commit_ == 0) {
+void Buffer::Commit() {
+	if (start_ == 0) {
 		return;
 	}
-	memmove(buf_, &buf_[commit_], len_ - commit_);
-	len_ -= commit_;
-	start_ -= commit_;
-	commit_ = 0;
+	memmove(buf_, &buf_[start_], len_ - start_);
+	len_ -= start_;
+	start_ = 0;
 }

buffer.h

@@ -8,23 +8,19 @@ class ConstBuffer {
 	ConstBuffer(const char *buf, size_t len);
 
 	[[nodiscard]] size_t ReadMaxLen() const;
-	[[nodiscard]] const char *Read(size_t len);
+	[[nodiscard]] virtual const char *Read(size_t len);
 	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 {
   public:
-  	Buffer(const char *buf, size_t len) = delete;
 	Buffer(char *buf, size_t size, size_t len);
   	Buffer(size_t size);
 
@@ -32,9 +28,9 @@ class Buffer : public ConstBuffer {
 	[[nodiscard]] size_t WriteMaxLen() const;
 	void Wrote(size_t len);
 
-	void Consume(); // discard up to last commit
+	void Commit(); // commit read position
 
-  private:
+  protected:
   	std::unique_ptr<char> own_buf_;
 	char *buf_;
 	const size_t size_;

example_simple.cpp

@@ -10,7 +10,6 @@ int main(int argc, char *argv[]) {
 	gflags::ParseCommandLineFlags(&argc, &argv, true);
 
 	FastCGIServer server(FLAGS_port, [](std::unique_ptr<FastCGIRequest> request) {
-		LOG(INFO) << "request from " << request->GetParam("REMOTE_ADDR");
 		request->Write({{"Content-Type", "text/plain"}}, {"Hello world"});
 		request->WriteEnd();
 	});

fastcgi_conn.cpp

@@ -55,7 +55,7 @@ constexpr auto fcgi_max_record_len = sizeof(fcgi_header) + fcgi_max_content_len
 FastCGIConn::FastCGIConn(int sock, const sockaddr_in6& client_addr, const std::function<void(std::unique_ptr<FastCGIRequest>)>& callback)
 		: sock_(sock),
 		  callback_(callback),
-		  buf_(fcgi_max_record_len) {
+		  buf_(sock, fcgi_max_record_len) {
 	char client_addr_str[INET6_ADDRSTRLEN];
 	PCHECK(inet_ntop(AF_INET6, &client_addr.sin6_addr, client_addr_str, sizeof(client_addr_str)));
 
@@ -67,22 +67,6 @@ FastCGIConn::~FastCGIConn() {
 	LOG(INFO) << "connection closed";
 }
 
-void FastCGIConn::Serve() {
-	while (true) {
-		auto read_len = read(sock_, buf_.WritePtr(), buf_.WriteMaxLen());
-		PCHECK(read_len >= 0);
-		if (read_len == 0) {
-			LOG(INFO) << "peer closed connection";
-			delete this;
-			return;
-		}
-		buf_.Wrote(read_len);
-
-		ParseBuf();
-		buf_.Consume(); // free buffer tail space for next read
-	}
-}
-
 void FastCGIConn::WriteBlock(uint8_t type, uint16_t request_id, const std::vector<iovec>& vecs) {
 	std::vector<iovec> out_vecs;
 	out_vecs.reserve(vecs.size() + 1);
@@ -123,17 +107,16 @@ void FastCGIConn::WriteEnd(uint16_t request_id) {
 	WriteBlock(3, request_id, vecs);
 }
 
-void FastCGIConn::ParseBuf() {
-	buf_.ResetRead();
-
+void FastCGIConn::Serve() {
 	while (true) {
 		const auto *header = buf_.ReadObj<fcgi_header>();
 		if (!header) {
-			return;
+			LOG(INFO) << "readobj failed";
+			break;
 		}
 
 		CHECK_EQ(header->version, 1);
-		if (buf_.ReadMaxLen() < header->ContentLength() + header->padding_length) {
+		if (!buf_.BufferAtLeast(header->ContentLength())) {
 			return;
 		}
 
@@ -170,10 +153,19 @@ void FastCGIConn::ParseBuf() {
 					request_->AddIn(in);
 				}
 			}
+			break;
+
+		  default:
+			CHECK(false) << "unknown record type: " << header->type;
+			break;
 		}
 
-		CHECK(buf_.Discard(header->padding_length));
+		if (!buf_.Discard(header->padding_length)) {
+			break;
+		}
 
 		buf_.Commit(); // we've acted on the bytes read so far
 	}
+
+	delete this;
 }

fastcgi_conn.h

@@ -3,7 +3,7 @@
 #include <functional>
 #include <unordered_map>
 
-#include "buffer.h"
+#include "stream_buffer.h"
 
 struct sockaddr_in6;
 class FastCGIRequest;
@@ -20,12 +20,10 @@ class FastCGIConn {
 	void WriteEnd(uint16_t request_id);
 
   private:
-  	void ParseBuf();
-
   	const int sock_;
   	std::function<void(std::unique_ptr<FastCGIRequest>)> callback_;
 
-	Buffer buf_;
+	StreamBuffer buf_;
 
 	std::unique_ptr<FastCGIRequest> request_;
 };

stream_buffer.cpp

@@ -0,0 +1,26 @@
+#include "stream_buffer.h"
+
+StreamBuffer::StreamBuffer(int sock, size_t size)
+		: Buffer(size),
+		  sock_(sock) {}
+
+bool StreamBuffer::BufferAtLeast(size_t len) {
+	CHECK_LE(start_ + len, size_);
+
+	while (ReadMaxLen() < len) {
+		auto read_len = read(sock_, WritePtr(), WriteMaxLen());
+		PCHECK(read_len >= 0);
+		if (read_len == 0) {
+			return false;
+		}
+		Wrote(read_len);
+	}
+	return true;
+}
+
+const char *StreamBuffer::Read(size_t len) {
+	if (!BufferAtLeast(len)) {
+		return nullptr;
+	}
+	return Buffer::Read(len);
+}

stream_buffer.h

@@ -0,0 +1,19 @@
+#pragma once
+
+#include "buffer.h"
+
+class StreamBuffer : public Buffer {
+  public:
+	StreamBuffer(int sock, size_t size);
+
+	[[nodiscard]] bool BufferAtLeast(size_t len);
+	[[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)));
+}