encoded_buffer documentation

split async_pipe into stdout and stdin, there's nothing in common

src/core/pipe.cpp

@@ -9,7 +9,7 @@ namespace mob
 static std::atomic<int> g_next_pipe_id(0);
 
 
-async_pipe::async_pipe(const context& cx)
+async_pipe_stdout::async_pipe_stdout(const context& cx)
 	: cx_(cx), pending_(false), closed_(true)
 {
 	buffer_ = std::make_unique<char[]>(buffer_size);
@@ -18,25 +18,15 @@ async_pipe::async_pipe(const context& cx)
 	std::memset(&ov_, 0, sizeof(ov_));
 }
 
-bool async_pipe::closed() const
+bool async_pipe_stdout::closed() const
 {
 	return closed_;
 }
 
-handle_ptr async_pipe::create_for_stdout()
-{
-	return create(true);
-}
-
-handle_ptr async_pipe::create_for_stdin()
-{
-	return create(false);
-}
-
-handle_ptr async_pipe::create(bool for_stdout)
+handle_ptr async_pipe_stdout::create()
 {
 	// creating pipe
-	handle_ptr out(for_stdout ? create_named_pipe() : create_anonymous_pipe());
+	handle_ptr out(create_named_pipe());
 	if (out.get() == INVALID_HANDLE_VALUE)
 		return {};
 
@@ -55,7 +45,7 @@ handle_ptr async_pipe::create(bool for_stdout)
 	return out;
 }
 
-std::string_view async_pipe::read(bool finish)
+std::string_view async_pipe_stdout::read(bool finish)
 {
 	std::string_view s;
 
@@ -76,19 +66,7 @@ std::string_view async_pipe::read(bool finish)
 	return s;
 }
 
-std::size_t async_pipe::write(std::string_view s)
-{
-	const DWORD n = static_cast<DWORD>(s.size());
-	DWORD written = 0;
-	const auto r = ::WriteFile(stdout_.get(), s.data(), n, &written, nullptr);
-
-	if (written >= s.size())
-		stdout_ = {};
-
-	return written;
-}
-
-HANDLE async_pipe::create_named_pipe()
+HANDLE async_pipe_stdout::create_named_pipe()
 {
 	const auto pipe_id = g_next_pipe_id.fetch_add(1) + 1;
 
@@ -152,35 +130,7 @@ HANDLE async_pipe::create_named_pipe()
 	return output_write;
 }
 
-HANDLE async_pipe::create_anonymous_pipe()
-{
-	SECURITY_ATTRIBUTES saAttr = {};
-	saAttr.nLength = sizeof(SECURITY_ATTRIBUTES);
-	saAttr.bInheritHandle = TRUE;
-
-	// Create a pipe for the child process's STDIN.
-	HANDLE read_pipe, write_pipe;
-	if (!CreatePipe(&read_pipe, &write_pipe, &saAttr, 0))
-	{
-		const auto e = GetLastError();
-		cx_.bail_out(context::cmd,
-			"CreatePipe failed, {}", error_message(e));
-	}
-
-	// Ensure the write handle to the pipe for STDIN is not inherited.
-	if (!SetHandleInformation(write_pipe, HANDLE_FLAG_INHERIT, 0))
-	{
-		const auto e = GetLastError();
-		cx_.bail_out(context::cmd,
-			"SetHandleInformation failed, {}", error_message(e));
-	}
-
-	stdout_.reset(write_pipe);
-
-	return read_pipe;
-}
-
-std::string_view async_pipe::try_read()
+std::string_view async_pipe_stdout::try_read()
 {
 	DWORD bytes_read = 0;
 
@@ -219,7 +169,7 @@ std::string_view async_pipe::try_read()
 	return {buffer_.get(), bytes_read};
 }
 
-std::string_view async_pipe::check_pending()
+std::string_view async_pipe_stdout::check_pending()
 {
 	DWORD bytes_read = 0;
 
@@ -275,4 +225,60 @@ std::string_view async_pipe::check_pending()
 	return {buffer_.get(), bytes_read};
 }
 
+
+async_pipe_stdin::async_pipe_stdin(const context& cx)
+	: cx_(cx)
+{
+}
+
+handle_ptr async_pipe_stdin::create()
+{
+	// creating pipe
+	handle_ptr out(create_anonymous_pipe());
+	if (out.get() == INVALID_HANDLE_VALUE)
+		return {};
+
+	return out;
+}
+
+std::size_t async_pipe_stdin::write(std::string_view s)
+{
+	const DWORD n = static_cast<DWORD>(s.size());
+	DWORD written = 0;
+	const auto r = ::WriteFile(stdin_.get(), s.data(), n, &written, nullptr);
+
+	if (written >= s.size())
+		stdin_ = {};
+
+	return written;
+}
+
+HANDLE async_pipe_stdin::create_anonymous_pipe()
+{
+	SECURITY_ATTRIBUTES saAttr = {};
+	saAttr.nLength = sizeof(SECURITY_ATTRIBUTES);
+	saAttr.bInheritHandle = TRUE;
+
+	// Create a pipe for the child process's STDIN.
+	HANDLE read_pipe, write_pipe;
+	if (!CreatePipe(&read_pipe, &write_pipe, &saAttr, 0))
+	{
+		const auto e = GetLastError();
+		cx_.bail_out(context::cmd,
+			"CreatePipe failed, {}", error_message(e));
+	}
+
+	// Ensure the write handle to the pipe for STDIN is not inherited.
+	if (!SetHandleInformation(write_pipe, HANDLE_FLAG_INHERIT, 0))
+	{
+		const auto e = GetLastError();
+		cx_.bail_out(context::cmd,
+			"SetHandleInformation failed, {}", error_message(e));
+	}
+
+	stdin_.reset(write_pipe);
+
+	return read_pipe;
+}
+
 }	// namespace

src/core/pipe.h

@@ -5,17 +5,14 @@
 namespace mob
 {
 
-class async_pipe
+class async_pipe_stdout
 {
 public:
-	async_pipe(const context& cx);
+	async_pipe_stdout(const context& cx);
 
-	handle_ptr create_for_stdout();
+	handle_ptr create();
 	std::string_view read(bool finish);
 
-	handle_ptr create_for_stdin();
-	std::size_t write(std::string_view s);
-
 	bool closed() const;
 
 private:
@@ -29,12 +26,26 @@ private:
 	bool pending_;
 	bool closed_;
 
-	handle_ptr create(bool for_stdout);
 	HANDLE create_named_pipe();
-	HANDLE create_anonymous_pipe();
 
 	std::string_view try_read();
 	std::string_view check_pending();
 };
 
+
+class async_pipe_stdin
+{
+public:
+	async_pipe_stdin(const context& cx);
+
+	handle_ptr create();
+	std::size_t write(std::string_view s);
+
+private:
+	const context& cx_;
+	handle_ptr stdin_;
+
+	HANDLE create_anonymous_pipe();
+};
+
 }	// namespace

src/core/process.cpp

@@ -281,15 +281,15 @@ void process::do_run(const std::string& what)
 
 	handle_ptr stdout_pipe, stderr_pipe;
 
-	impl_.stdout_pipe.reset(new async_pipe(*cx_));
-	impl_.stderr_pipe.reset(new async_pipe(*cx_));
+	impl_.stdout_pipe.reset(new async_pipe_stdout(*cx_));
+	impl_.stderr_pipe.reset(new async_pipe_stdout(*cx_));
 
 	switch (stdout_.flags)
 	{
 		case forward_to_log:
 		case keep_in_string:
 		{
-			stdout_pipe = impl_.stdout_pipe->create_for_stdout();
+			stdout_pipe = impl_.stdout_pipe->create();
 			si.hStdOutput = stdout_pipe.get();
 			break;
 		}
@@ -312,7 +312,7 @@ void process::do_run(const std::string& what)
 		case forward_to_log:
 		case keep_in_string:
 		{
-			stderr_pipe = impl_.stderr_pipe->create_for_stdout();
+			stderr_pipe = impl_.stderr_pipe->create();
 			si.hStdError = stderr_pipe.get();
 			break;
 		}
@@ -332,8 +332,8 @@ void process::do_run(const std::string& what)
 
 	if (stdin_)
 	{
-		impl_.stdin_pipe.reset(new async_pipe(*cx_));
-		impl_.stdin_handle = impl_.stdin_pipe->create_for_stdin();
+		impl_.stdin_pipe.reset(new async_pipe_stdin(*cx_));
+		impl_.stdin_handle = impl_.stdin_pipe->create();
 	}
 	else
 	{
@@ -454,7 +454,7 @@ void process::read_pipes(bool finish)
 }
 
 void process::read_pipe(
-	bool finish, stream& s, async_pipe& pipe, context::reason r)
+	bool finish, stream& s, async_pipe_stdout& pipe, context::reason r)
 {
 	switch (s.flags)
 	{
@@ -768,20 +768,4 @@ std::string process::arg_to_string(int i, arg_flags)
 	return std::to_string(i);
 }
 
-
-encoded_buffer::encoded_buffer(encodings e, std::string bytes)
-	: e_(e), bytes_(std::move(bytes)), last_(0)
-{
-}
-
-void encoded_buffer::add(std::string_view bytes)
-{
-	bytes_.append(bytes.begin(), bytes.end());
-}
-
-std::string encoded_buffer::utf8_string() const
-{
-	return bytes_to_utf8(e_, bytes_);
-}
-
 }	// namespace

src/core/process.h

@@ -8,7 +8,8 @@ namespace mob
 {
 
 class url;
-class async_pipe;
+class async_pipe_stdout;
+class async_pipe_stdin;
 
 class process
 {
@@ -173,9 +174,9 @@ private:
 		handle_ptr handle;
 		handle_ptr job;
 		std::atomic<bool> interrupt{false};
-		std::unique_ptr<async_pipe> stdout_pipe;
-		std::unique_ptr<async_pipe> stderr_pipe;
-		std::unique_ptr<async_pipe> stdin_pipe;
+		std::unique_ptr<async_pipe_stdout> stdout_pipe;
+		std::unique_ptr<async_pipe_stdout> stderr_pipe;
+		std::unique_ptr<async_pipe_stdin> stdin_pipe;
 		handle_ptr stdin_handle;
 
 		impl() = default;
@@ -225,7 +226,7 @@ private:
 
 	void do_run(const std::string& what);
 	void read_pipes(bool finish);
-	void read_pipe(bool finish, stream& s, async_pipe& pipe, context::reason r);
+	void read_pipe(bool finish, stream& s, async_pipe_stdout& pipe, context::reason r);
 
 	void on_completed();
 	void on_timeout(bool& already_interrupted);

src/utility/string.cpp

@@ -442,4 +442,20 @@ std::string path_to_utf8(fs::path p)
 	return utf16_to_utf8(p.native());
 }
 
+
+encoded_buffer::encoded_buffer(encodings e, std::string bytes)
+	: e_(e), bytes_(std::move(bytes)), last_(0)
+{
+}
+
+void encoded_buffer::add(std::string_view bytes)
+{
+	bytes_.append(bytes.begin(), bytes.end());
+}
+
+std::string encoded_buffer::utf8_string() const
+{
+	return bytes_to_utf8(e_, bytes_);
+}
+
 } // namespace

src/utility/string.h

@@ -167,18 +167,50 @@ void for_each_line(std::string_view s, F&& f)
 }
 
 
+// an array of bytes using the specified encoding that can be parsed to call a
+// function with every line
+//
+// the output of a process is stored in an encoded_buffer and next_utf8_lines()
+// is called to process every line in it, avoiding copies or memory allocation,
+// except for conversions to utf8 when necessary
+//
+// if the encoding is dont_know, the buffer is basically interpreted as ascii
+// for checking newlines and the bytes are given as-is to the callback
+//
 class encoded_buffer
 {
 public:
+	// a buffer using the given encoding and starting bytes
+	//
 	encoded_buffer(encodings e=encodings::dont_know, std::string bytes={});
 
+	// copies bytes to the internal buffer
+	//
 	void add(std::string_view bytes);
 
+	// returns a copy of the internal buffer as utf8
+	//
 	std::string utf8_string() const;
 
+	// calls `f()` with a utf8 string for every non-empty line in the buffer;
+	// remembers the final offset when next_utf8_lines() was last called so
+	// lines are only processed once
+	//
+	// if `finished` is false, it's assumed that more bytes will arrive
+	// eventually, so the bytes after the last newline in the buffer are not
+	// considered a line
+	//
+	// if `finished` is true, it's assumed that the output is complete and that
+	// the final bytes before the end of the buffer are considered a valid line
+	//
 	template <class F>
 	void next_utf8_lines(bool finished, F&& f)
 	{
+		// every case is the same, except for conversions:
+		//   1) get the next, non-empty line
+		//   2) if the line is empty, there wasn't any, so break
+		//   3) convert to utf8 if needed and call f()
+
 		for (;;)
 		{
 			switch (e_)
@@ -226,40 +258,79 @@ public:
 	}
 
 private:
+	// encoding of the buffer
 	encodings e_;
+
+	// internal buffer
 	std::string bytes_;
+
+	// offset of the last newline found the last time next_utf8_lines() was
+	// called
 	std::size_t last_;
 
+
+	// looks for the next newline character after last_ and returns a
+	// string_view of the data between the two; empty lines are ignored,
+	// handles both lf and crlf the same
+	//
+	// this is a static function, but it's always given bytes_ and last_ as
+	// arguments
+	//
 	template <class CharT>
-	std::basic_string_view<CharT> next_line(
-		bool finished, std::string_view bytes, std::size_t& byte_offset)
+	static std::basic_string_view<CharT> next_line(
+		bool finished, const std::string_view bytes, std::size_t& byte_offset)
 	{
-		std::size_t size = bytes.size();
-
-		if constexpr (sizeof(CharT) == 2)
+		// number of available bytes in the buffer
+		//
+		// for utf16, it's possible (but unlikely) that the buffer has an odd
+		// number of bytes if not all the output was flushed, so don't check the
+		// last stray byte
+		//
+		// this doesn't handle stray bytes for other encodings, but they use
+		// single bytes for cr and lf, so it's fine
+		//
+		const std::size_t size = [&]
 		{
-			if ((size & 1) == 1)
-				--size;
-		}
+			if constexpr (sizeof(CharT) == 2)
+			{
+				if ((bytes.size() & 1) == 1)
+					return bytes.size() - 1;
+			}
 
+			return bytes.size();
+		}();
+
+
+		// position just past where the last newline was found
 		const CharT* start = reinterpret_cast<const CharT*>(bytes.data() + byte_offset);
-		const CharT* end = reinterpret_cast<const CharT*>(bytes.data() + size);
+
+		// end of the buffer
+		const CharT* const end = reinterpret_cast<const CharT*>(bytes.data() + size);
+
+		// current character being checked
 		const CharT* p = start;
 
+		// line that was found, or empty if none is available
 		std::basic_string_view<CharT> line;
 
+
+		// looking for a non-empty line
 		while (p != end)
 		{
 			if (*p == CharT('\n') || *p == CharT('\r'))
 			{
 				line = {start, static_cast<std::size_t>(p - start)};
 
+				// skip newline characters from this point
 				while (p != end && (*p == CharT('\n') || *p == CharT('\r')))
 					++p;
 
+				// line is not empty, take it
 				if (!line.empty())
 					break;
 
+				// line can be empty for something like \n\n, continue looking
+				// for a non-empty line if that's the case
 				start = p;
 			}
 			else
@@ -268,20 +339,30 @@ private:
 			}
 		}
 
+		// if the line is empty but `finished` is true, make sure the last
+		// line in the buffer is handled
+		//
 		if (line.empty())
 		{
 			if (finished)
 			{
+				// the line is from past the last newline to the end of the
+				// buffer; this may be empty if the buffer actually ends with
+				// a newline, which is fine
 				line = {
 					reinterpret_cast<const CharT*>(bytes.data() + byte_offset),
 					size - byte_offset
 				};
 
+				// tell the caller that whole thing has been processed
 				byte_offset = bytes.size();
 			}
 		}
 		else
 		{
+			// a non-empty line was found, update the offset to be past the
+			// newline character(s)
+
 			byte_offset = static_cast<std::size_t>(
 				reinterpret_cast<const char*>(p) - bytes.data());