Finish implementing the multipart protocol, now multi file uploads work without corruption.

Common/Net/Sinks.cpp

@@ -9,6 +9,7 @@
 #include "Common/Net/Sinks.h"
 
 #include "Common/Log.h"
+#include "Common/StringUtils.h"
 #include "Common/File/FileDescriptor.h"
 
 #ifndef MSG_NOSIGNAL
@@ -44,10 +45,41 @@ bool InputSink::ReadLineWithEnding(std::string &s) {
 		memcpy(&s[0], buf_ + read_, newline + 1);
 	}
 	AccountDrain(newline + 1);
-
 	return true;
 }
 
+std::pair<std::string_view, std::string_view> InputSink::BufferParts() const {
+	if (read_ + valid_ <= BUFFER_SIZE) {
+		return {std::string_view(buf_ + read_, valid_), std::string_view()};
+	} else {
+		size_t firstPartSize = BUFFER_SIZE - read_;
+		size_t secondPartSize = valid_ - firstPartSize;
+		return {std::string_view(buf_ + read_, firstPartSize), std::string_view(buf_, secondPartSize)};
+	}
+}
+
+size_t InputSink::ReadBinaryUntilTerminator(char *dest, size_t bufSize, std::string_view terminator, bool *didReadTerminator) {
+	Block();
+
+	auto [part1, part2] = BufferParts();
+	size_t offset = SplitSearch(terminator, part1, part2);
+	if (offset == std::string_view::npos) {
+		*didReadTerminator = false;
+		// Not found, read as much as we can - but leave space for the terminator
+		const s64 toRead = std::min((s64)valid_, (s64)bufSize - (s64)terminator.length());
+		TakeExact(dest, toRead);
+		return toRead;
+	} else {
+		// Terminator found! Read right up to it, and then skip it.
+		*didReadTerminator = true;
+		_dbg_assert_(offset < valid_);
+		TakeExact(dest, offset);
+		Skip(terminator.size());
+		_dbg_assert_(valid_ >= 0);
+		return offset;
+	}
+}
+
 std::string InputSink::ReadLineWithEnding() {
 	std::string s;
 	ReadLineWithEnding(s);

Common/Net/Sinks.h

@@ -21,20 +21,36 @@ public:
 	bool Skip(size_t bytes);
 	void Discard();
 
+	size_t ReadBinaryUntilTerminator(char *dest, size_t bufSize, std::string_view terminator, bool *didReadTerminator);
+
 	bool Empty() const;
 	bool TryFill();
 
+	size_t ValidAmount() const {
+		return valid_;
+	}
+
+	// Size of the internal buffer. Useful for some clients.
+	enum {
+		BUFFER_SIZE = 32 * 1024,
+	};
+
 private:
+	std::pair<std::string_view, std::string_view> BufferParts() const;
 	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_;
+
+	// Circular buffer. read_ is the position to read from, write_ is the position to write to,
+	// valid_ is the number of valid bytes. valid_ can wrap around, so you might need to take
+	// two segments when reading and split when writing.
+
 	char buf_[BUFFER_SIZE];
 	size_t read_;
 	size_t write_;

Common/StringUtils.cpp

@@ -557,3 +557,27 @@ void MakeUnique(std::vector<std::string> &v) {
 	}
 	v.swap(result);
 }
+
+size_t SplitSearch(std::string_view needle, std::string_view part1, std::string_view part2) {
+	if (part1.find(needle) != std::string_view::npos) {
+		// Easy case, found in part1.
+		return part1.find(needle);
+	}
+	size_t part1Size = part1.size();
+	size_t maxOverlap = std::min(needle.size() - 1, part1Size);
+	for (size_t overlap = maxOverlap; overlap > 0; overlap--) {
+		if (part1.substr(part1Size - overlap) == needle.substr(0, overlap)) {
+			// Found an overlap.
+			size_t remaining = needle.size() - overlap;
+			if (part2.substr(0, remaining) == needle.substr(overlap)) {
+				return part1Size - overlap;
+			}
+		}
+	}
+	// Now, check if it's found in part2 instead.
+	size_t posInPart2 = part2.find(needle);
+	if (posInPart2 != std::string_view::npos) {
+		return part1Size + posInPart2;
+	}
+	return std::string_view::npos;
+}

Common/StringUtils.h

@@ -154,6 +154,8 @@ inline void CharArrayFromFormat(char (& out)[Count], const char* format, ...)
 
 void MakeUnique(std::vector<std::string> &vec);
 
+size_t SplitSearch(std::string_view needle, std::string_view part1, std::string_view part2);
+
 // Replaces %1, %2, %3 in format with arg1, arg2, arg3.
 // Much safer than snprintf and friends.
 // For mixes of strings and ints, manually convert the ints to strings.

Common/TimeUtil.h

@@ -37,6 +37,7 @@ void GetCurrentTimeFormatted(char formattedTime[13]);
 // Most accurate timer possible - no extra double conversions. Only for spans.
 class Instant {
 public:
+	Instant();
 	static Instant Now() {
 		return Instant();
 	}
@@ -44,7 +45,6 @@ public:
 	double ElapsedMs() const { return ElapsedSeconds() * 1000.0; }
 	int64_t ElapsedNanos() const;
 private:
-	Instant();
 	uint64_t nativeStart_;
 #ifndef _WIN32
 	int64_t nsecs_;

Core/WebServer.cpp

@@ -55,9 +55,21 @@ static WebServerFlags serverFlags;
 
 std::mutex g_webServerLock;
 static Path g_uploadPath;  // TODO: Supply this through registration instead.
+static std::atomic<int> g_uploadSessionId(1);
+std::map<int, UploadProgress> g_uploadsInProgress;
 
 // NOTE: These *only* encode spaces, which is almost enough.
 
+std::vector<UploadProgress> GetUploadsInProgress() {
+	std::lock_guard<std::mutex> guard(g_webServerLock);
+	std::vector<UploadProgress> uploads;
+	uploads.reserve(g_uploadsInProgress.size());
+	for (const auto &pair : g_uploadsInProgress) {
+		uploads.push_back(pair.second);
+	}
+	return uploads;
+}
+
 std::string ServerUriEncode(std::string_view plain) {
 	return ReplaceAll(plain, " ", "%20");
 }
@@ -444,51 +456,40 @@ static void HandleUploadUI(const http::ServerRequest &request) {
 		"</body></html>");
 }
 
-static void HandleUploadPost(const http::ServerRequest &request) {
-	AndroidJNIThreadContext jniContext;
-
-	// Do some sanity checks.
-	if (request.Method() != http::RequestHeader::POST) {
-		ERROR_LOG(Log::HTTP, "Wrong method");
-		return;
-	}
-
-	Path uploadPath;
-	{
+class ProgressTracker {
+public:
+	ProgressTracker(int sessionId) : sessionId_(sessionId) {
 		std::lock_guard<std::mutex> guard(g_webServerLock);
-		uploadPath = g_uploadPath;
+		g_uploadsInProgress[sessionId_] = UploadProgress();
 	}
+	~ProgressTracker() {
+		std::lock_guard<std::mutex> guard(g_webServerLock);
+		g_uploadsInProgress.erase(sessionId_);
+	}
+private:
+	int sessionId_;
+};
 
-	// Now start handling things.
-	std::string contentType;
-	if (!request.GetHeader("content-type", &contentType)) {
-		return;
-	}
-	size_t bpos = contentType.find("boundary=");
-	if (bpos == std::string::npos) {
-		return;
-	}
-	const std::string boundary = contentType.substr(bpos + strlen("boundary="));
-
-	// The total length of the entire multipart thing.
-	u64 contentLength = request.Header().content_length;
-	if (contentLength == 0) {
-		WARN_LOG(Log::HTTP, "Bad content length");
-		return;
-	}
+enum class MultiPartResult {
+	MoveToNext,
+	RequestError,
+	LocalError,
+	Done,
+};
 
+static MultiPartResult HandleMultipartPart(const http::ServerRequest &request, std::string boundary, const Path &uploadPath, ProgressTracker &progress) {
 	std::string firstBoundary = request.In()->ReadLine();
 	if (firstBoundary != "--" + boundary) {
-		WARN_LOG(Log::HTTP, "Bad boundary: Expected --%s but got %s", boundary);
-		return;
+		WARN_LOG(Log::HTTP, "Bad boundary: Expected --%s but got %s", boundary.c_str());
+		return MultiPartResult::RequestError;
 	}
 
 	std::string disposition = request.In()->ReadLine();
 	std::vector<std::string_view> parts;
 	SplitString(disposition, ';', parts);
 	if (parts.size() < 2 || !startsWith(parts[0], "Content-Disposition: form-data")) {
-		WARN_LOG(Log::HTTP, "Bad content disposition: %s", disposition);
-		return;
+		WARN_LOG(Log::HTTP, "Bad content disposition: %s", disposition.c_str());
+		return MultiPartResult::RequestError;
 	}
 
 	std::string filename;
@@ -514,7 +515,7 @@ static void HandleUploadPost(const http::ServerRequest &request) {
 
 	if (filename.empty()) {
 		ERROR_LOG(Log::HTTP, "Didn't receive a filename");
-		return;
+		return MultiPartResult::RequestError;
 	}
 
 	std::string fileContentType = request.In()->ReadLine();
@@ -522,60 +523,150 @@ static void HandleUploadPost(const http::ServerRequest &request) {
 
 	Path destPath = uploadPath / filename;
 
+	bool dryRun = false;
 	if (File::Exists(destPath)) {
-		INFO_LOG(Log::HTTP, "File already exists: %s", destPath.ToVisualString().c_str());
-		return;
+		INFO_LOG(Log::HTTP, "File already exists, entering dry run mode: %s", destPath.ToVisualString().c_str());
+		dryRun = true;
 	}
 
 	// Make sure the destination exists.
 	File::CreateFullPath(destPath.NavigateUp());
 
-	INFO_LOG(Log::HTTP, "Receiving '%s', writing to '%s' (%d bytes)...", filename.c_str(), destPath.ToVisualString().c_str());
+	INFO_LOG(Log::HTTP, "Receiving '%s', writing to '%s' (unknown number of bytes)...", filename.c_str(), destPath.ToVisualString().c_str());
 
 	// OK, enter a loop where we read some data until we hit the boundary again.
 	// The boundary is chosen to be "unique" and unlikely to appear in the file. We trust that.
-	Buffer buffer;
-	FILE *fp = File::OpenCFile(destPath, "wb");
-	if (!fp) {
-		ERROR_LOG(Log::HTTP, "Failed to open destination file '%s' for writing", destPath.ToVisualString().c_str());
-		return;
+	FILE *fp = nullptr;
+	if (!dryRun) {
+		fp = File::OpenCFile(destPath, "wb");
+		if (!fp) {
+			ERROR_LOG(Log::HTTP, "Failed to open destination file '%s' for writing, entering dry run mode.", destPath.ToVisualString().c_str());
+			dryRun = true;  // We still want to keep reading the input stream - there might be more files following.
+		}
 	}
-	u64 bytesWritten = 0;
+
+	u64 bytesTransferred = 0;
+	char buffer[net::InputSink::BUFFER_SIZE];
 	while (true) {
-		// NOTE: Lines here can be extremely long, especially in compressed data. So we should split ReadLine up if needed.
-		std::string line = request.In()->ReadLine();
-		if (line == "--" + boundary || line == "--" + boundary + "--") {
-			INFO_LOG(Log::HTTP, "Line matches boundary, breaking.");
-			// Done.
+		bool terminatorFound = false;
+		size_t readBytes = request.In()->ReadBinaryUntilTerminator(buffer, sizeof(buffer), "\r\n--" + boundary, &terminatorFound);
+		if (fp) {
+			if (fwrite(buffer, 1, readBytes, fp) != readBytes) {
+				ERROR_LOG(Log::HTTP, "Failed to write %d bytes to destination file '%s' - disk full?", (int)readBytes, destPath.ToVisualString().c_str());
+				fclose(fp);
+				return MultiPartResult::RequestError;
+			}
+		}
+		bytesTransferred += readBytes;
+		if (terminatorFound) {
+			INFO_LOG(Log::HTTP, "Found terminator, skipping and proceeding.");
 			break;
 		}
-		// Write the line and a newline (since we ate it). TODO: This could be avoided.
-		size_t len = line.size();
-		if (fwrite(line.data(), 1, len, fp) != len || fwrite("\r\n", 1, 2, fp) != 2) {
-			ERROR_LOG(Log::HTTP, "Failed to write %d bytes to destination file '%s' - bailing", (int)len, destPath.ToVisualString().c_str());
-			fclose(fp);
-			return;
-		}
-		bytesWritten += line.size();
 	}
 
-	INFO_LOG(Log::HTTP, "Total bytes written: %d", (int)bytesWritten);
-	fclose(fp);
+	INFO_LOG(Log::HTTP, "Total bytes transferred: %d", (int)bytesTransferred);
+	if (fp) {
+		fclose(fp);
+	}
 
 	// NOTE: We already read the boundary above.
+	// However if this is the last part, the boundary will have "--\r\n" after it, otherwise there will be a line break.
+	// So, let's read two more bytes, to see if it's "--" or "\r\n".
+	std::string ending(2, 'x');
+	request.In()->TakeExact(ending.data(), 2);
+
+	if (ending == "--") {
+		INFO_LOG(Log::HTTP, "Upload of '%s' complete, '--' encountered. %d bytes left in buffer.", filename.c_str(), (int)request.In()->ValidAmount());
+		// Read the final \r\n.
+		std::string finalCRLF;
+		finalCRLF.resize(2);
+		request.In()->TakeExact(finalCRLF.data(), 2);
+		if (finalCRLF != "\r\n") {
+			// Not a big deal, but log it.
+			WARN_LOG(Log::HTTP, "Expected final CRLF after ending '--', got '%02x %02x'", finalCRLF[0], finalCRLF[1]);
+		}
+		return MultiPartResult::Done;
+	} else if (ending == "\r\n") {
+		INFO_LOG(Log::HTTP, "Upload of '%s' complete, continuing to next part.", filename.c_str());
+		return MultiPartResult::MoveToNext;
+	} else {
+		WARN_LOG(Log::HTTP, "Unexpected upload ending: '%s'. Bailing.", ending.c_str());
+		return MultiPartResult::RequestError;
+	}
+}
+
+// Handles a POST to upload a file.
+// This uses the HTTP multipart protocol, which is arcane and complicated, unfortunately.
+static void HandleUploadPost(const http::ServerRequest &request) {
+	AndroidJNIThreadContext jniContext;
+
+	// Do some sanity checks.
+	if (request.Method() != http::RequestHeader::POST) {
+		ERROR_LOG(Log::HTTP, "Wrong method");
+		return;
+	}
+
+	Path uploadPath;
+	{
+		std::lock_guard<std::mutex> guard(g_webServerLock);
+		uploadPath = g_uploadPath;
+	}
+
+	if (uploadPath.empty()) {
+		ERROR_LOG(Log::HTTP, "HandleUploadPost should not happen while uploadPath is empty");
+		return;
+	}
+
+	// Now start handling things.
+	std::string contentType;
+	if (!request.GetHeader("content-type", &contentType)) {
+		return;
+	}
+	size_t bpos = contentType.find("boundary=");
+	if (bpos == std::string::npos) {
+		return;
+	}
+	const std::string boundary = contentType.substr(bpos + strlen("boundary="));
+
+	// The total length of the entire multipart thing.
+	u64 contentLength = request.Header().content_length;
+	if (contentLength == 0) {
+		WARN_LOG(Log::HTTP, "Bad content length");
+		return;
+	}
+
+	const int sessionId = g_uploadSessionId.fetch_add(1);
+
+	ProgressTracker progress(sessionId);
+	while (true) {
+		MultiPartResult result = HandleMultipartPart(request, boundary, uploadPath, progress);
+		switch (result) {
+		case MultiPartResult::Done:
+			goto exitLoop;
+		case MultiPartResult::RequestError:
+			request.WriteHttpResponseHeader("1.0", 400, -1, "text/plain");  // Bad request
+			return;
+		case MultiPartResult::LocalError:
+			request.WriteHttpResponseHeader("1.0", 500, -1, "text/plain");  // Server error
+			// Disk full etc.
+			return;
+		case MultiPartResult::MoveToNext:
+			// Else just continue to the next part.
+			break;
+		}
+	}
+exitLoop:
 
 	// Now the buffer should be empty.
 	if (!request.In()->Empty()) {
-		WARN_LOG(Log::HTTP, "We didn't drain the request.");
-		std::string extraLine = request.In()->ReadLine();
-		INFO_LOG(Log::HTTP, "Extra line: %s", extraLine.c_str());
+		size_t remaining = request.In()->ValidAmount();
+		WARN_LOG(Log::HTTP, "We didn't fully drain the request (%d bytes still buffered)", (int)remaining);
+		// std::string extraBytes;
+		// extraBytes.resize(remaining);
+		// request.In()->TakeExact(extraBytes.data(), remaining);
 	}
 
-	INFO_LOG(Log::HTTP, "Upload of '%s' complete.", filename.c_str());
-	// request.WriteHttpResponseHeader("1.0", 200, -1, "text/plain");
-
-	const size_t blockSize = 16 * 1024;
-
+	request.WriteHttpResponseHeader("1.0", 200, -1, "text/plain");
 }
 
 void WebServerSetUploadPath(const Path &path) {

Core/WebServer.h

@@ -17,6 +17,7 @@
 
 #pragma once
 
+#include <vector>
 #include "Common/Common.h"
 
 class Path;
@@ -41,3 +42,12 @@ void ShutdownWebServer();
 bool RemoteISOFileSupported(const std::string &filename);
 void WebServerSetUploadPath(const Path &path);
 int WebServerPort();
+
+struct UploadProgress {
+	size_t uploadedBytes = 0;
+	size_t totalBytes = 0;
+	size_t uploadedFiles = 0;  // we don't know the count ahead of time
+	std::string currentFilename;
+};
+
+std::vector<UploadProgress> GetUploadsInProgress();

UI/MainScreen.cpp

@@ -1584,6 +1584,9 @@ void MainScreen::dialogFinished(const Screen *dialog, DialogResult result) {
 	} else if (tag == "IAP") {
 		// Gold status may have changed.
 		RecreateViews();
+	} else if (tag == "Upload") {
+		// Files may have been uploaded.
+		RecreateViews();
 	}
 }
 

UI/UploadScreen.cpp

@@ -1,5 +1,6 @@
 #include "Common/Net/Resolve.h"
 #include "Common/System/Request.h"
+#include "Common/File/DiskFree.h"
 #include "UI/UploadScreen.h"
 #include "Common/StringUtils.h"
 #include "Core/WebServer.h"
@@ -38,6 +39,16 @@ void UploadScreen::CreateViews() {
 
 	root_->Add(new TextView(std::string(co->T("Uploading to: ")) + targetFolder_.ToVisualString()));
 
+	// Show information about current upload streams (there can be multiple, but normally it's just one
+	// where files are uploaded sequentially).
+	std::vector<UploadProgress> uploads = GetUploadsInProgress();
+	for (const auto &upload : uploads) {
+		std::string uploadText = StringFromFormat("%zu / %zu bytes uploaded (%zu files). Current file: %s",
+			upload.uploadedBytes, upload.totalBytes, upload.uploadedFiles,
+			upload.currentFilename.c_str());
+		root_->Add(new TextView(uploadText));
+	}
+
 	//infoText->SetTextAlignment(TEXT_ALIGN_CENTER);
 	//verticalLayout->AddView(infoText);
 	//AddStandardBack(verticalLayout);
@@ -51,4 +62,10 @@ void UploadScreen::update() {
 		prevRunning_ = running;
 		RecreateViews();
 	}
+
+	// Update in-progress uploads approximately every half second.
+	if (lastUpdate_.ElapsedSeconds() > 0.5) {
+		RecreateViews();
+		lastUpdate_ = Instant::Now();
+	}
 }

UI/UploadScreen.h

@@ -26,6 +26,7 @@
 #include "Common/UI/UIScreen.h"
 #include "Common/UI/ViewGroup.h"
 #include "UI/MiscScreens.h"
+#include "Common/TimeUtil.h"
 
 // Upload screen: Shows the user an ip address to go to in a web browser on the same network,
 // in order to upload game files to the current directory.
@@ -45,4 +46,5 @@ private:
 	bool prevRunning_ = false;
 	std::vector<std::string> localIPs_;
 	Path targetFolder_;
+	Instant lastUpdate_;
 };

Windows/WASAPIContext.cpp

@@ -12,6 +12,7 @@
 
 #include "Common/Data/Encoding/Utf8.h"
 #include "Common/Log.h"
+#include "Common/Thread/ThreadUtil.h"
 #include "WASAPIContext.h"
 
 using Microsoft::WRL::ComPtr;
@@ -308,6 +309,8 @@ void WASAPIContext::FrameUpdate(bool allowAutoChange) {
 }
 
 void WASAPIContext::AudioLoop() {
+	SetCurrentThreadName("WASAPIAudioLoop");
+
 	DWORD taskID = 0;
 	HANDLE mmcssHandle = nullptr;
 	if (latencyMode_ == LatencyMode::Aggressive) {
@@ -323,7 +326,7 @@ void WASAPIContext::AudioLoop() {
 		audioClient_->GetBufferSize(&available);
 	}
 
-	AudioFormat format = Classify(format_);
+	const AudioFormat format = Classify(format_);
 	const int nChannels = format_->nChannels;
 
 	while (running_) {

unittest/UnitTest.cpp

@@ -1232,6 +1232,23 @@ bool TestVolumeFunc() {
 	return true;
 }
 
+bool TestSplitSearch() {
+	std::string part1 = "The quick brown fox jumps";
+	std::string part2 = " over the lazy dog.";
+
+	size_t offset = SplitSearch("jumps over", part1, part2);
+	EXPECT_EQ_INT(offset, 20);
+	offset = SplitSearch("quick", part1, part2);
+	EXPECT_EQ_INT(offset, 4);
+	offset = SplitSearch(" over", part1, part2);
+	EXPECT_EQ_INT(offset, 25);
+	offset = SplitSearch("fox jumps", part1, part2);
+	EXPECT_EQ_INT(offset, 16);
+	offset = SplitSearch("dog.", part1, part2);
+	EXPECT_EQ_INT(offset, 40);
+	return true;
+}
+
 typedef bool (*TestFunc)();
 struct TestItem {
 	const char *name;
@@ -1301,6 +1318,7 @@ TestItem availableTests[] = {
 	TEST_ITEM(SIMD),
 	TEST_ITEM(CrossSIMD),
 	TEST_ITEM(VolumeFunc),
+	TEST_ITEM(SplitSearch),
 };
 
 int main(int argc, const char *argv[]) {