http: Switch to using input/output sinks.

This should be more performant than reading one byte at a time in a loop.

CMakeLists.txt

@@ -1001,6 +1001,8 @@ add_library(native STATIC
 	ext/native/net/http_client.h
 	ext/native/net/resolve.cpp
 	ext/native/net/resolve.h
+	ext/native/net/sinks.cpp
+	ext/native/net/sinks.h
 	ext/native/net/url.cpp
 	ext/native/net/url.h
 	ext/native/profiler/profiler.cpp

ext/native/Android.mk

@@ -62,6 +62,7 @@ LOCAL_SRC_FILES :=\
     net/http_server.cpp \
     net/http_headers.cpp \
     net/resolve.cpp \
+    net/sinks.cpp \
     net/url.cpp \
     profiler/profiler.cpp \
     thread/executor.cpp \

ext/native/native.vcxproj

@@ -264,6 +264,7 @@
     <ClInclude Include="net\resolve.h" />
     <ClInclude Include="net\url.h" />
     <ClInclude Include="profiler\profiler.h" />
+    <ClInclude Include="net\sinks.h" />
     <ClInclude Include="thin3d\d3dx9_loader.h" />
     <ClInclude Include="thin3d\thin3d.h" />
     <ClInclude Include="thread\executor.h" />
@@ -723,6 +724,7 @@
     <ClCompile Include="net\resolve.cpp" />
     <ClCompile Include="net\url.cpp" />
     <ClCompile Include="profiler\profiler.cpp" />
+    <ClCompile Include="net\sinks.cpp" />
     <ClCompile Include="thin3d\d3dx9_loader.cpp" />
     <ClCompile Include="thin3d\thin3d.cpp" />
     <ClCompile Include="thin3d\thin3d_d3d9.cpp" />

ext/native/native.vcxproj.filters

@@ -308,6 +308,9 @@
     <ClInclude Include="math\dataconv.h">
       <Filter>math</Filter>
     </ClInclude>
+    <ClInclude Include="net\sinks.h">
+      <Filter>net</Filter>
+    </ClInclude>
   </ItemGroup>
   <ItemGroup>
     <ClCompile Include="gfx\gl_debug_log.cpp">
@@ -745,6 +748,9 @@
     <ClCompile Include="thin3d\thin3d_vulkan.cpp">
       <Filter>thin3d</Filter>
     </ClCompile>
+    <ClCompile Include="net\sinks.cpp">
+      <Filter>net</Filter>
+    </ClCompile>
   </ItemGroup>
   <ItemGroup>
     <Filter Include="gfx">

ext/native/net/http_headers.cpp

@@ -6,6 +6,7 @@
 #include "base/logging.h"
 #include "base/stringutil.h"
 #include "file/fd_util.h"
+#include "net/sinks.h"
 
 namespace http {
 
@@ -121,30 +122,26 @@ int RequestHeader::ParseHttpHeader(const char *buffer) {
   return 0;
 }
 
-void RequestHeader::ParseHeaders(int fd) {
-  int line_count = 0;
-  // Loop through request headers.
-  while (true) {
-    if (!fd_util::WaitUntilReady(fd, 5.0)) {  // Wait max 5 secs.
-      // Timed out or error.
-      ok = false;
-      return;
-    }
-    char buffer[1024];
-    fd_util::ReadLine(fd, buffer, 1023);
-    StringTrimEndNonAlphaNum(buffer);
-    if (buffer[0] == '\0')
-      break;
-    ParseHttpHeader(buffer);
-    line_count++;
-    if (type == SIMPLE) {
-      // Done!
-      ILOG("Simple: Done parsing http request.");
-      break;
-    }
-  }
-  ILOG("finished parsing request.");
-  ok = line_count > 1;
+void RequestHeader::ParseHeaders(net::InputSink *sink) {
+	int line_count = 0;
+	std::string line;
+	while (sink->ReadLine(line)) {
+		if (line.length() == 0) {
+			// Blank line, this means end of headers.
+			break;
+		}
+
+		ParseHttpHeader(line.c_str());
+		line_count++;
+		if (type == SIMPLE) {
+			// Done!
+			ILOG("Simple: Done parsing http request.");
+			break;
+		}
+	}
+
+	ILOG("finished parsing request.");
+	ok = line_count > 1;
 }
 
 }  // namespace http

ext/native/net/http_headers.h

@@ -3,6 +3,10 @@
 
 #include "base/buffer.h"
 
+namespace net {
+class InputSink;
+};
+
 namespace http {
 
 class RequestHeader {
@@ -12,6 +16,7 @@ class RequestHeader {
   // Public variables since it doesn't make sense
   // to bother with accessors for all these.
   int status;
+  // Intentional misspelling.
   char *referer;
   char *user_agent;
   char *resource;
@@ -29,7 +34,7 @@ class RequestHeader {
   };
   Method method;
   bool ok;
-  void ParseHeaders(int fd);
+  void ParseHeaders(net::InputSink *sink);
   bool GetParamValue(const char *param_name, std::string *value) const;
  private:
   int ParseHttpHeader(const char *buffer);

ext/native/net/http_server.cpp

@@ -2,6 +2,7 @@
 
 #ifdef _WIN32
 
+#define NOMINMAX
 #include <winsock2.h>
 #include <ws2tcpip.h>
 #include <io.h>
@@ -15,8 +16,11 @@
 #include <arpa/inet.h>        /*  inet (3) funtions         */
 #include <unistd.h>           /*  misc. UNIX functions      */
 
+#define closesocket close
+
 #endif
 
+#include <algorithm>
 #include <stdio.h>
 #include <stdlib.h>
 
@@ -25,50 +29,48 @@
 #include "base/buffer.h"
 #include "file/fd_util.h"
 #include "net/http_server.h"
+#include "net/sinks.h"
 #include "thread/executor.h"
 
 namespace http {
 
 Request::Request(int fd)
     : fd_(fd) {
-  in_buffer_ = new Buffer;
-  out_buffer_ = new Buffer;
-  header_.ParseHeaders(fd_);
+	in_ = new net::InputSink(fd);
+	out_ = new net::OutputSink(fd);
+	header_.ParseHeaders(in_);
 
-  if (header_.ok) {
-    // Read the rest, too.
-    if (header_.content_length >= 0) {
-      in_buffer_->Read(fd_, header_.content_length);
-    }
-    ILOG("The request carried with it %i bytes", (int)in_buffer_->size());
-  } else {
-    Close();
-  }
+	if (header_.ok) {
+		ILOG("The request carried with it %i bytes", (int)header_.content_length);
+	} else {
+	    Close();
+	}
 }
 
 Request::~Request() {
   Close();
 
-  CHECK(in_buffer_->empty());
-  delete in_buffer_;
-  CHECK(out_buffer_->empty());
-  delete out_buffer_;
+  CHECK(in_->Empty());
+  delete in_;
+  CHECK(out_->Empty());
+  delete out_;
 }
 
 void Request::WriteHttpResponseHeader(int status, int size) const {
-  Buffer *buffer = out_buffer_;
+  net::OutputSink *buffer = Out();
   buffer->Printf("HTTP/1.0 %d OK\r\n", status);
-  buffer->Append("Server: SuperDuperServer v0.1\r\n");
-  buffer->Append("Content-Type: text/html\r\n");
+  buffer->Push("Server: SuperDuperServer v0.1\r\n");
+  buffer->Push("Content-Type: text/html\r\n");
+  buffer->Push("Connection: close\r\n");
   if (size >= 0) {
     buffer->Printf("Content-Length: %i\r\n", size);
   }
-  buffer->Append("\r\n");
+  buffer->Push("\r\n");
 }
 
 void Request::WritePartial() const {
   CHECK(fd_);
-  out_buffer_->Flush(fd_);
+  out_->Flush();
 }
 
 void Request::Write() {
@@ -79,7 +81,7 @@ void Request::Write() {
 
 void Request::Close() {
   if (fd_) {
-    close(fd_);
+    closesocket(fd_);
     fd_ = 0;
   }
 }
@@ -139,7 +141,12 @@ void Server::HandleConnection(int conn_fd) {
     return;
   }
   HandleRequestDefault(request);
-  request.WritePartial();
+
+  // TODO: Way to mark the content body as read, read it here if never read.
+  // This allows the handler to stream if need be.
+
+  // TODO: Could handle keep alive here.
+  request.Write();
 }
 
 void Server::HandleRequest(const Request &request) {
@@ -157,12 +164,13 @@ void Server::HandleRequestDefault(const Request &request) {
   ILOG("No handler for '%s', falling back to 404.", request.resource());
   const char *payload = "<html><body>404 not found</body></html>\r\n";
   request.WriteHttpResponseHeader(404, (int)strlen(payload));
-  request.out_buffer()->Append(payload);
+  request.Out()->Push(payload);
 }
 
 void Server::HandleListing(const Request &request) {
+  request.WriteHttpResponseHeader(200, -1);
   for (auto iter = handlers_.begin(); iter != handlers_.end(); ++iter) {
-    request.out_buffer()->Printf("%s", iter->first.c_str());
+    request.Out()->Printf("%s\n", iter->first.c_str());
   }
 }
 

ext/native/net/http_server.h

@@ -8,6 +8,11 @@
 #include "net/http_headers.h"
 #include "thread/executor.h"
 
+namespace net {
+class InputSink;
+class OutputSink;
+};
+
 namespace http {
 
 class Request {
@@ -23,8 +28,8 @@ class Request {
     return header_.GetParamValue(param_name, value);
   }
 
-  Buffer *in_buffer() const { return in_buffer_; }
-  Buffer *out_buffer() const { return out_buffer_; }
+  net::InputSink *In() const { return in_; }
+  net::OutputSink *Out() const { return out_; }
 
   // TODO: Remove, in favor of PartialWrite and friends.
   int fd() const { return fd_; }
@@ -38,11 +43,11 @@ class Request {
   // If size is negative, no Content-Length: line is written.
   void WriteHttpResponseHeader(int status, int size = -1) const;
 
- private:
-  Buffer *in_buffer_;
-  Buffer *out_buffer_;
-  RequestHeader header_;
-  int fd_;
+private:
+	net::InputSink *in_;
+	net::OutputSink *out_;
+	RequestHeader header_;
+	int fd_;
 };
 
 // Register handlers on this class to serve stuff.

ext/native/net/sinks.cpp

@@ -0,0 +1,364 @@
+#pragma optimize("", off)
+
+#ifdef _WIN32
+
+#define NOMINMAX
+#include <winsock2.h>
+#include <ws2tcpip.h>
+#include <io.h>
+
+#else
+
+#include <sys/socket.h>       /*  socket definitions        */
+#include <sys/types.h>        /*  socket types              */
+#include <sys/wait.h>         /*  for waitpid()             */
+#include <netinet/in.h>       /*  struct sockaddr_in        */
+#include <arpa/inet.h>        /*  inet (3) funtions         */
+#include <unistd.h>           /*  misc. UNIX functions      */
+
+#endif
+
+#include <algorithm>
+#include <cstdarg>
+
+#include "base/logging.h"
+#include "net/sinks.h"
+#include "file/fd_util.h"
+
+namespace net {
+
+InputSink::InputSink(size_t fd) : fd_(fd), read_(0), write_(0), valid_(0) {
+	fd_util::SetNonBlocking((int)fd_, true);
+}
+
+bool InputSink::ReadLineWithEnding(std::string &s) {
+	size_t newline = FindNewline();
+	if (newline == BUFFER_SIZE) {
+		Block();
+		newline = FindNewline();
+	}
+	if (newline == BUFFER_SIZE) {
+		// Timed out.
+		return false;
+	}
+
+	s.resize(newline + 1);
+	if (read_ + newline + 1 > BUFFER_SIZE) {
+		// Need to do two reads.
+		size_t chunk1 = BUFFER_SIZE - read_;
+		size_t chunk2 = read_ + newline + 1 - BUFFER_SIZE;
+		memcpy(&s[0], buf_ + read_, chunk1);
+		memcpy(&s[chunk1], buf_, chunk2);
+	} else {
+		memcpy(&s[0], buf_ + read_, newline + 1);
+	}
+	AccountDrain(newline + 1);
+
+	return true;
+}
+
+std::string InputSink::ReadLineWithEnding() {
+	std::string s;
+	ReadLineWithEnding(s);
+	return s;
+}
+
+bool InputSink::ReadLine(std::string &s) {
+	bool result = ReadLineWithEnding(s);
+	if (result) {
+		size_t l = s.length();
+		if (l >= 2 && s[l - 2] == '\r' && s[l - 1] == '\n') {
+			s.resize(l - 2);
+		} else if (l >= 1 && s[l - 1] == '\n') {
+			s.resize(l - 1);
+		}
+	}
+	return result;
+}
+
+std::string InputSink::ReadLine() {
+	std::string s;
+	ReadLine(s);
+	return s;
+}
+
+size_t InputSink::FindNewline() const {
+	// Technically, \r\n, but most parsers are lax... let's follow suit.
+	size_t until_end = std::min(valid_, BUFFER_SIZE - read_);
+	for (const char *p = buf_ + read_, *end = buf_ + read_ + until_end; p < end; ++p) {
+		if (*p == '\n') {
+			return p - (buf_ + read_);
+		}
+	}
+
+	// Were there more bytes wrapped around?
+	if (read_ + valid_ > BUFFER_SIZE) {
+		size_t wrapped = read_ + valid_ - BUFFER_SIZE;
+		for (const char *p = buf_, *end = buf_ + wrapped; p < end; ++p) {
+			if (*p == '\n') {
+				// Offset by the skipped portion before wrapping.
+				return (p - buf_) + until_end;
+			}
+		}
+	}
+
+	// Never found, return an invalid position to indicate.
+	return BUFFER_SIZE;
+}
+
+bool InputSink::TakeExact(char *buf, size_t bytes) {
+	while (bytes > 0) {
+		size_t drained = TakeAtMost(buf, bytes);
+		buf += drained;
+		bytes -= drained;
+
+		if (drained == 0) {
+			if (!Block()) {
+				// Timed out reading more bytes.
+				return false;
+			}
+		}
+	}
+
+	return true;
+}
+
+size_t InputSink::TakeAtMost(char *buf, size_t bytes) {
+	Fill();
+
+	// The least of: contiguous to read, actually populated in buffer, and wanted.
+	size_t avail = std::min(std::min(BUFFER_SIZE - read_, valid_), bytes);
+
+	if (avail != 0) {
+		memcpy(buf, buf_ + read_, avail);
+		AccountDrain(avail);
+	}
+
+	return avail;
+}
+
+bool InputSink::Skip(size_t bytes) {
+	while (bytes > 0) {
+		size_t drained = std::min(valid_, bytes);
+		AccountDrain(drained);
+		bytes -= drained;
+
+		// Nothing left to read?  Get more.
+		if (drained == 0) {
+			if (!Block()) {
+				// Timed out reading more bytes.
+				return false;
+			}
+		}
+	}
+
+	return true;
+}
+
+void InputSink::Fill() {
+	// Avoid small reads if possible.
+	if (BUFFER_SIZE - valid_ > PRESSURE) {
+		// Whatever isn't valid and follows write_ is what's available.
+		size_t avail = BUFFER_SIZE - std::max(write_, valid_);
+
+		int bytes = recv(fd_, buf_ + write_, (int)avail, 0);
+		AccountFill(bytes);
+	}
+}
+
+bool InputSink::Block() {
+	if (!fd_util::WaitUntilReady((int)fd_, 5.0)) {
+		return false;
+	}
+
+	Fill();
+	return true;
+}
+
+void InputSink::AccountFill(int bytes) {
+	if (bytes < 0) {
+		ELOG("Error reading from socket");
+		return;
+	}
+
+	// Okay, move forward (might be by zero.)
+	valid_ += bytes;
+	write_ += bytes;
+	if (write_ >= BUFFER_SIZE) {
+		write_ -= BUFFER_SIZE;
+	}
+}
+
+void InputSink::AccountDrain(size_t bytes) {
+	valid_ -= bytes;
+	read_ += bytes;
+	if (read_ >= BUFFER_SIZE) {
+		read_ -= BUFFER_SIZE;
+	}
+}
+
+bool InputSink::Empty() {
+	return valid_ == 0;
+}
+
+OutputSink::OutputSink(size_t fd) : fd_(fd), read_(0), write_(0), valid_(0) {
+	fd_util::SetNonBlocking((int)fd_, true);
+}
+
+bool OutputSink::Push(const std::string &s) {
+	return Push(&s[0], s.length());
+}
+
+bool OutputSink::Push(const char *buf, size_t bytes) {
+	while (bytes > 0) {
+		size_t pushed = PushAtMost(buf, bytes);
+		buf += pushed;
+		bytes -= pushed;
+
+		if (pushed == 0) {
+			if (!Block()) {
+				// We couldn't write all the bytes.
+				return false;
+			}
+		}
+	}
+
+	return true;
+}
+
+bool OutputSink::PushCRLF(const std::string &s) {
+	if (Push(s)) {
+		return Push("r\n", 2);
+	}
+	return false;
+}
+
+size_t OutputSink::PushAtMost(const char *buf, size_t bytes) {
+	Drain();
+
+	// Look for contiguous free space after write_ that's valid.
+	size_t avail = std::min(BUFFER_SIZE - std::max(write_, valid_), bytes);
+
+	if (avail != 0) {
+		memcpy(buf_ + write_, buf, avail);
+		AccountPush(avail);
+	}
+
+	return avail;
+}
+
+
+bool OutputSink::Printf(const char *fmt, ...) {
+	// Let's start by checking how much space we have.
+	size_t avail = BUFFER_SIZE - std::max(write_, valid_);
+
+	va_list args;
+	va_start(args, fmt);
+	// Make a backup in case we don't have sufficient space.
+	va_list backup;
+	va_copy(backup, args);
+
+	bool success = true;
+
+	int result = vsnprintf(buf_ + write_, avail, fmt, args);
+	if (result >= avail) {
+		// There wasn't enough space.  Let's use a buffer instead.
+		// This could be caused by wraparound.
+		char temp[BUFFER_SIZE];
+		result = vsnprintf(temp, BUFFER_SIZE, fmt, args);
+
+		if (result < BUFFER_SIZE && result > 0) {
+			// In case it did return the null terminator.
+			if (temp[result - 1] == '\0') {
+				result--;
+			}
+
+			success = Push(temp, result);
+			// We've written so there's nothing more.
+			result = 0;
+		}
+	}
+	va_end(args);
+	va_end(backup);
+
+	// Okay, did we actually write?
+	if (result >= avail) {
+		// This means the result string was too big for the buffer.
+		ELOG("Not enough space to format output.");
+		return false;
+	} else if (result < 0) {
+		ELOG("vsnprintf failed.");
+		return false;
+	}
+
+	if (result > 0) {
+		AccountPush(result);
+	}
+
+	return success;
+}
+
+bool OutputSink::Block() {
+	if (!fd_util::WaitUntilReady((int)fd_, 5.0, true)) {
+		return false;
+	}
+
+	Drain();
+	return true;
+}
+
+bool OutputSink::Flush() {
+	while (valid_ > 0) {
+		size_t avail = std::min(BUFFER_SIZE - read_, valid_);
+
+		int bytes = send(fd_, buf_ + read_, (int)avail, 0);
+		AccountDrain(bytes);
+
+		if (bytes == 0) {
+			// This may also drain.  Either way, keep looping.
+			if (!Block()) {
+				return false;
+			}
+		}
+	}
+
+	return true;
+}
+
+void OutputSink::Drain() {
+	// Avoid small reads if possible.
+	if (valid_ > PRESSURE) {
+		// Let's just do contiguous valid.
+		size_t avail = std::min(BUFFER_SIZE - read_, valid_);
+
+		int bytes = send(fd_, buf_ + read_, (int)avail, 0);
+		AccountDrain(bytes);
+	}
+}
+
+void OutputSink::AccountPush(size_t bytes) {
+	valid_ += bytes;
+	write_ += bytes;
+	if (write_ >= BUFFER_SIZE) {
+		write_ -= BUFFER_SIZE;
+	}
+}
+
+void OutputSink::AccountDrain(int bytes) {
+	if (bytes < 0) {
+		ELOG("Error writing to socket");
+		return;
+	}
+
+	valid_ -= bytes;
+	read_ += bytes;
+	if (read_ >= BUFFER_SIZE) {
+		read_ -= BUFFER_SIZE;
+	}
+}
+
+bool OutputSink::Empty() {
+	return valid_ == 0;
+}
+
+};

ext/native/net/sinks.h

@@ -0,0 +1,72 @@
+#pragma once
+
+#include <string>
+
+namespace net {
+
+class InputSink {
+public:
+	InputSink(size_t fd);
+
+	bool ReadLine(std::string &s);
+	std::string ReadLine();
+	bool ReadLineWithEnding(std::string &s);
+	std::string ReadLineWithEnding();
+
+	// Read exactly this number of bytes, or fail.
+	bool TakeExact(char *buf, size_t bytes);
+	// Read whatever is convenient (may even return 0 bytes when there's more coming eventually.)
+	size_t TakeAtMost(char *buf, size_t bytes);
+	// Skip exactly this number of bytes, or fail.
+	bool Skip(size_t bytes);
+
+	bool Empty();
+
+private:
+	void Fill();
+	bool Block();
+	void AccountFill(int bytes);
+	void AccountDrain(size_t bytes);
+	size_t FindNewline() const;
+
+	static const size_t BUFFER_SIZE = 32 * 1024;
+	static const size_t PRESSURE = 8 * 1024;
+
+	size_t fd_;
+	char buf_[BUFFER_SIZE];
+	size_t read_;
+	size_t write_;
+	size_t valid_;
+};
+
+class OutputSink {
+public:
+	OutputSink(size_t fd);
+
+	bool Push(const std::string &s);
+	bool Push(const char *buf, size_t bytes);
+	size_t PushAtMost(const char *buf, size_t bytes);
+	bool PushCRLF(const std::string &s);
+	bool Printf(const char *fmt, ...);
+
+	bool Flush();
+
+	bool Empty();
+
+private:
+	void Drain();
+	bool Block();
+	void AccountPush(size_t bytes);
+	void AccountDrain(int bytes);
+
+	static const size_t BUFFER_SIZE = 32 * 1024;
+	static const size_t PRESSURE = 8 * 1024;
+
+	size_t fd_;
+	char buf_[BUFFER_SIZE];
+	size_t read_;
+	size_t write_;
+	size_t valid_;
+};
+
+};