move file i/o related files to separate file and remove duplicated code (#576)

1 files changed, 595 insertions, 0 deletions

diff --git a/src/zenremotestore/filesystemutils.cpp b/src/zenremotestore/filesystemutils.cpp
new file mode 100644
index 000000000..20ab3faea
--- /dev/null
+++ b/src/zenremotestore/filesystemutils.cpp
@@ -0,0 +1,595 @@
+// Copyright Epic Games, Inc. All Rights Reserved.
+
+#include "filesystemutils.h"
+
+#include <zenremotestore/chunking/chunkedcontent.h>
+
+#include <zencore/filesystem.h>
+#include <zencore/fmtutils.h>
+#include <zencore/parallelwork.h>
+#include <zencore/scopeguard.h>
+#include <zencore/timer.h>
+#include <zencore/trace.h>
+
+namespace zen {
+
+BufferedOpenFile::BufferedOpenFile(const std::filesystem::path Path,
+								   std::atomic<uint64_t>&	   OpenReadCount,
+								   std::atomic<uint64_t>&	   CurrentOpenFileCount,
+								   std::atomic<uint64_t>&	   ReadCount,
+								   std::atomic<uint64_t>&	   ReadByteCount)
+: m_Source(Path, BasicFile::Mode::kRead)
+, m_SourceSize(m_Source.FileSize())
+, m_OpenReadCount(OpenReadCount)
+, m_CurrentOpenFileCount(CurrentOpenFileCount)
+, m_ReadCount(ReadCount)
+, m_ReadByteCount(ReadByteCount)
+
+{
+	m_OpenReadCount++;
+	m_CurrentOpenFileCount++;
+}
+
+BufferedOpenFile::~BufferedOpenFile()
+{
+	m_CurrentOpenFileCount--;
+}
+
+CompositeBuffer
+BufferedOpenFile::GetRange(uint64_t Offset, uint64_t Size)
+{
+	ZEN_TRACE_CPU("BufferedOpenFile::GetRange");
+
+	ZEN_ASSERT((m_CacheBlockIndex == (uint64_t)-1) || m_Cache);
+	auto _ = MakeGuard([&]() { ZEN_ASSERT((m_CacheBlockIndex == (uint64_t)-1) || m_Cache); });
+
+	ZEN_ASSERT((Offset + Size) <= m_SourceSize);
+	const uint64_t BlockIndexStart = Offset / BlockSize;
+	const uint64_t BlockIndexEnd   = (Offset + Size - 1) / BlockSize;
+
+	std::vector<SharedBuffer> BufferRanges;
+	BufferRanges.reserve(BlockIndexEnd - BlockIndexStart + 1);
+
+	uint64_t ReadOffset = Offset;
+	for (uint64_t BlockIndex = BlockIndexStart; BlockIndex <= BlockIndexEnd; BlockIndex++)
+	{
+		const uint64_t BlockStartOffset = BlockIndex * BlockSize;
+		if (m_CacheBlockIndex != BlockIndex)
+		{
+			uint64_t CacheSize = Min(BlockSize, m_SourceSize - BlockStartOffset);
+			ZEN_ASSERT(CacheSize > 0);
+			m_Cache = IoBuffer(CacheSize);
+			m_Source.Read(m_Cache.GetMutableView().GetData(), CacheSize, BlockStartOffset);
+			m_ReadCount++;
+			m_ReadByteCount += CacheSize;
+			m_CacheBlockIndex = BlockIndex;
+		}
+
+		const uint64_t BytesRead = ReadOffset - Offset;
+		ZEN_ASSERT(BlockStartOffset <= ReadOffset);
+		const uint64_t OffsetIntoBlock = ReadOffset - BlockStartOffset;
+		ZEN_ASSERT(OffsetIntoBlock < m_Cache.GetSize());
+		const uint64_t BlockBytes = Min(m_Cache.GetSize() - OffsetIntoBlock, Size - BytesRead);
+		BufferRanges.emplace_back(SharedBuffer(IoBuffer(m_Cache, OffsetIntoBlock, BlockBytes)));
+		ReadOffset += BlockBytes;
+	}
+	CompositeBuffer Result(std::move(BufferRanges));
+	ZEN_ASSERT(Result.GetSize() == Size);
+	return Result;
+}
+
+ReadFileCache::ReadFileCache(std::atomic<uint64_t>&		  OpenReadCount,
+							 std::atomic<uint64_t>&		  CurrentOpenFileCount,
+							 std::atomic<uint64_t>&		  ReadCount,
+							 std::atomic<uint64_t>&		  ReadByteCount,
+							 const std::filesystem::path& Path,
+							 const ChunkedFolderContent&  LocalContent,
+							 const ChunkedContentLookup&  LocalLookup,
+							 size_t						  MaxOpenFileCount)
+: m_Path(Path)
+, m_LocalContent(LocalContent)
+, m_LocalLookup(LocalLookup)
+, m_OpenReadCount(OpenReadCount)
+, m_CurrentOpenFileCount(CurrentOpenFileCount)
+, m_ReadCount(ReadCount)
+, m_ReadByteCount(ReadByteCount)
+{
+	m_OpenFiles.reserve(MaxOpenFileCount);
+}
+ReadFileCache::~ReadFileCache()
+{
+	m_OpenFiles.clear();
+}
+
+CompositeBuffer
+ReadFileCache::GetRange(uint32_t SequenceIndex, uint64_t Offset, uint64_t Size)
+{
+	ZEN_TRACE_CPU("ReadFileCache::GetRange");
+
+	auto CacheIt =
+		std::find_if(m_OpenFiles.begin(), m_OpenFiles.end(), [SequenceIndex](const auto& Lhs) { return Lhs.first == SequenceIndex; });
+	if (CacheIt != m_OpenFiles.end())
+	{
+		if (CacheIt != m_OpenFiles.begin())
+		{
+			auto CachedFile(std::move(CacheIt->second));
+			m_OpenFiles.erase(CacheIt);
+			m_OpenFiles.insert(m_OpenFiles.begin(), std::make_pair(SequenceIndex, std::move(CachedFile)));
+		}
+		CompositeBuffer Result = m_OpenFiles.front().second->GetRange(Offset, Size);
+		return Result;
+	}
+	const uint32_t				LocalPathIndex = m_LocalLookup.SequenceIndexFirstPathIndex[SequenceIndex];
+	const std::filesystem::path LocalFilePath  = (m_Path / m_LocalContent.Paths[LocalPathIndex]).make_preferred();
+	if (Size == m_LocalContent.RawSizes[LocalPathIndex])
+	{
+		IoBuffer Result = IoBufferBuilder::MakeFromFile(LocalFilePath);
+		return CompositeBuffer(SharedBuffer(Result));
+	}
+	if (m_OpenFiles.size() == m_OpenFiles.capacity())
+	{
+		m_OpenFiles.pop_back();
+	}
+	m_OpenFiles.insert(
+		m_OpenFiles.begin(),
+		std::make_pair(
+			SequenceIndex,
+			std::make_unique<BufferedOpenFile>(LocalFilePath, m_OpenReadCount, m_CurrentOpenFileCount, m_ReadCount, m_ReadByteCount)));
+	CompositeBuffer Result = m_OpenFiles.front().second->GetRange(Offset, Size);
+	return Result;
+}
+
+uint32_t
+SetNativeFileAttributes(const std::filesystem::path FilePath, SourcePlatform SourcePlatform, uint32_t Attributes)
+{
+#if ZEN_PLATFORM_WINDOWS
+	if (SourcePlatform == SourcePlatform::Windows)
+	{
+		SetFileAttributesToPath(FilePath, Attributes);
+		return Attributes;
+	}
+	else
+	{
+		uint32_t CurrentAttributes = GetFileAttributesFromPath(FilePath);
+		uint32_t NewAttributes	   = zen::MakeFileAttributeReadOnly(CurrentAttributes, zen::IsFileModeReadOnly(Attributes));
+		if (CurrentAttributes != NewAttributes)
+		{
+			SetFileAttributesToPath(FilePath, NewAttributes);
+		}
+		return NewAttributes;
+	}
+#endif	// ZEN_PLATFORM_WINDOWS
+#if ZEN_PLATFORM_LINUX || ZEN_PLATFORM_MAC
+	if (SourcePlatform != SourcePlatform::Windows)
+	{
+		zen::SetFileMode(FilePath, Attributes);
+		return Attributes;
+	}
+	else
+	{
+		uint32_t CurrentMode = zen::GetFileMode(FilePath);
+		uint32_t NewMode	 = zen::MakeFileModeReadOnly(CurrentMode, zen::IsFileAttributeReadOnly(Attributes));
+		if (CurrentMode != NewMode)
+		{
+			zen::SetFileMode(FilePath, NewMode);
+		}
+		return NewMode;
+	}
+#endif	// ZEN_PLATFORM_LINUX || ZEN_PLATFORM_MAC
+};
+
+uint32_t
+GetNativeFileAttributes(const std::filesystem::path FilePath)
+{
+#if ZEN_PLATFORM_WINDOWS
+	return GetFileAttributesFromPath(FilePath);
+#endif	// ZEN_PLATFORM_WINDOWS
+#if ZEN_PLATFORM_LINUX || ZEN_PLATFORM_MAC
+	return GetFileMode(FilePath);
+#endif	// ZEN_PLATFORM_LINUX || ZEN_PLATFORM_MAC
+}
+
+bool
+IsFileWithRetry(const std::filesystem::path& Path)
+{
+	std::error_code Ec;
+	bool			Result = IsFile(Path, Ec);
+	for (size_t Retries = 0; Ec && Retries < 3; Retries++)
+	{
+		Sleep(100 + int(Retries * 50));
+		Ec.clear();
+		Result = IsFile(Path, Ec);
+	}
+	if (Ec)
+	{
+		throw std::system_error(std::error_code(Ec.value(), std::system_category()),
+								fmt::format("Failed to check path '{}' is file, reason: ({}) {}", Path, Ec.value(), Ec.message()));
+	}
+	return Result;
+}
+
+bool
+SetFileReadOnlyWithRetry(const std::filesystem::path& Path, bool ReadOnly)
+{
+	std::error_code Ec;
+	bool			Result = SetFileReadOnly(Path, ReadOnly, Ec);
+	for (size_t Retries = 0; Ec && Retries < 3; Retries++)
+	{
+		Sleep(100 + int(Retries * 50));
+		if (!IsFileWithRetry(Path))
+		{
+			return false;
+		}
+		Ec.clear();
+		Result = SetFileReadOnly(Path, ReadOnly, Ec);
+	}
+	if (Ec)
+	{
+		throw std::system_error(std::error_code(Ec.value(), std::system_category()),
+								fmt::format("Failed {} read only flag for file '{}', reason: ({}) {}",
+											ReadOnly ? "setting" : "clearing",
+											Path,
+											Ec.value(),
+											Ec.message()));
+	}
+	return Result;
+}
+
+std::error_code
+RenameFileWithRetry(const std::filesystem::path& SourcePath, const std::filesystem::path& TargetPath)
+{
+	std::error_code Ec;
+	RenameFile(SourcePath, TargetPath, Ec);
+	for (size_t Retries = 0; Ec && Retries < 5; Retries++)
+	{
+		ZEN_ASSERT_SLOW(IsFile(SourcePath));
+		Sleep(50 + int(Retries * 150));
+		Ec.clear();
+		RenameFile(SourcePath, TargetPath, Ec);
+	}
+	return Ec;
+}
+
+std::error_code
+TryRemoveFile(const std::filesystem::path& Path)
+{
+	std::error_code Ec;
+	RemoveFile(Path, Ec);
+	if (Ec)
+	{
+		if (IsFile(Path, Ec))
+		{
+			Ec.clear();
+			RemoveFile(Path, Ec);
+			if (Ec)
+			{
+				return Ec;
+			}
+		}
+	}
+	return {};
+}
+
+void
+RemoveFileWithRetry(const std::filesystem::path& Path)
+{
+	std::error_code Ec;
+	RemoveFile(Path, Ec);
+	for (size_t Retries = 0; Ec && Retries < 6; Retries++)
+	{
+		Sleep(100 + int(Retries * 50));
+		if (!IsFileWithRetry(Path))
+		{
+			return;
+		}
+		Ec.clear();
+		RemoveFile(Path, Ec);
+	}
+	if (Ec)
+	{
+		throw std::system_error(std::error_code(Ec.value(), std::system_category()),
+								fmt::format("Failed removing file '{}', reason: ({}) {}", Path, Ec.value(), Ec.message()));
+	}
+}
+
+void
+FastCopyFile(bool						  AllowFileClone,
+			 bool						  UseSparseFiles,
+			 const std::filesystem::path& SourceFilePath,
+			 const std::filesystem::path& TargetFilePath,
+			 uint64_t					  RawSize,
+			 std::atomic<uint64_t>&		  WriteCount,
+			 std::atomic<uint64_t>&		  WriteByteCount,
+			 std::atomic<uint64_t>&		  CloneCount,
+			 std::atomic<uint64_t>&		  CloneByteCount)
+{
+	ZEN_TRACE_CPU("CopyFile");
+	if (AllowFileClone && TryCloneFile(SourceFilePath, TargetFilePath))
+	{
+		WriteCount += 1;
+		WriteByteCount += RawSize;
+		CloneCount += 1;
+		CloneByteCount += RawSize;
+	}
+	else
+	{
+		BasicFile TargetFile(TargetFilePath, BasicFile::Mode::kTruncate);
+		if (UseSparseFiles)
+		{
+			PrepareFileForScatteredWrite(TargetFile.Handle(), RawSize);
+		}
+		uint64_t Offset = 0;
+		if (!ScanFile(SourceFilePath, 512u * 1024u, [&](const void* Data, size_t Size) {
+				TargetFile.Write(Data, Size, Offset);
+				Offset += Size;
+				WriteCount++;
+				WriteByteCount += Size;
+			}))
+		{
+			throw std::runtime_error(fmt::format("Failed to copy file '{}' to '{}'", SourceFilePath, TargetFilePath));
+		}
+	}
+}
+
+CleanDirectoryResult
+CleanDirectory(
+	WorkerThreadPool&			 IOWorkerPool,
+	std::atomic<bool>&			 AbortFlag,
+	std::atomic<bool>&			 PauseFlag,
+	const std::filesystem::path& Path,
+	std::span<const std::string> ExcludeDirectories,
+	std::function<void(const std::string_view Details, uint64_t TotalCount, uint64_t RemainingCount, bool IsPaused, bool IsAborted)>&&
+			 ProgressFunc,
+	uint32_t ProgressUpdateDelayMS)
+{
+	ZEN_TRACE_CPU("CleanDirectory");
+	Stopwatch Timer;
+
+	std::atomic<uint64_t> DiscoveredItemCount = 0;
+	std::atomic<uint64_t> DeletedItemCount	  = 0;
+	std::atomic<uint64_t> DeletedByteCount	  = 0;
+
+	CleanDirectoryResult Result;
+	RwLock				 ResultLock;
+	auto				 _ = MakeGuard([&]() {
+		Result.DeletedCount		= DeletedItemCount.load();
+		Result.DeletedByteCount = DeletedByteCount.load();
+		Result.FoundCount		= DiscoveredItemCount.load();
+	});
+
+	ParallelWork Work(AbortFlag,
+					  PauseFlag,
+					  ProgressFunc ? WorkerThreadPool::EMode::DisableBacklog : WorkerThreadPool::EMode::EnableBacklog);
+
+	struct AsyncVisitor : public GetDirectoryContentVisitor
+	{
+		AsyncVisitor(const std::filesystem::path& InPath,
+					 std::atomic<bool>&			  InAbortFlag,
+					 std::atomic<uint64_t>&		  InDiscoveredItemCount,
+					 std::atomic<uint64_t>&		  InDeletedItemCount,
+					 std::atomic<uint64_t>&		  InDeletedByteCount,
+					 std::span<const std::string> InExcludeDirectories,
+					 CleanDirectoryResult&		  InResult,
+					 RwLock&					  InResultLock)
+		: Path(InPath)
+		, AbortFlag(InAbortFlag)
+		, DiscoveredItemCount(InDiscoveredItemCount)
+		, DeletedItemCount(InDeletedItemCount)
+		, DeletedByteCount(InDeletedByteCount)
+		, ExcludeDirectories(InExcludeDirectories)
+		, Result(InResult)
+		, ResultLock(InResultLock)
+		{
+		}
+		virtual void AsyncVisitDirectory(const std::filesystem::path& RelativeRoot, DirectoryContent&& Content) override
+		{
+			ZEN_TRACE_CPU("CleanDirectory_AsyncVisitDirectory");
+			if (!AbortFlag)
+			{
+				if (!Content.FileNames.empty())
+				{
+					DiscoveredItemCount += Content.FileNames.size();
+
+					const std::string RelativeRootString = RelativeRoot.generic_string();
+					bool			  RemoveContent		 = true;
+					for (const std::string_view ExcludeDirectory : ExcludeDirectories)
+					{
+						if (RelativeRootString.starts_with(ExcludeDirectory))
+						{
+							if (RelativeRootString.length() > ExcludeDirectory.length())
+							{
+								const char MaybePathDelimiter = RelativeRootString[ExcludeDirectory.length()];
+								if (MaybePathDelimiter == '/' || MaybePathDelimiter == '\\' ||
+									MaybePathDelimiter == std::filesystem::path::preferred_separator)
+								{
+									RemoveContent = false;
+									break;
+								}
+							}
+							else
+							{
+								RemoveContent = false;
+								break;
+							}
+						}
+					}
+					if (RemoveContent)
+					{
+						ZEN_TRACE_CPU("DeleteFiles");
+						for (size_t FileIndex = 0; FileIndex < Content.FileNames.size(); FileIndex++)
+						{
+							const std::filesystem::path& FileName = Content.FileNames[FileIndex];
+							const std::filesystem::path	 FilePath = (Path / RelativeRoot / FileName).make_preferred();
+
+							bool			IsRemoved = false;
+							std::error_code Ec;
+							(void)SetFileReadOnly(FilePath, false, Ec);
+							for (size_t Retries = 0; Ec && Retries < 3; Retries++)
+							{
+								Sleep(100 + int(Retries * 50));
+								if (!IsFileWithRetry(FilePath))
+								{
+									IsRemoved = true;
+									break;
+								}
+								Ec.clear();
+								(void)SetFileReadOnly(FilePath, false, Ec);
+							}
+							if (!IsRemoved && !Ec)
+							{
+								(void)RemoveFile(FilePath, Ec);
+								for (size_t Retries = 0; Ec && Retries < 6; Retries++)
+								{
+									Sleep(100 + int(Retries * 50));
+									if (!IsFileWithRetry(FilePath))
+									{
+										IsRemoved = true;
+										return;
+									}
+									Ec.clear();
+									(void)RemoveFile(FilePath, Ec);
+								}
+							}
+							if (!IsRemoved && Ec)
+							{
+								RwLock::ExclusiveLockScope _(ResultLock);
+								Result.FailedRemovePaths.push_back(std::make_pair(FilePath, Ec));
+							}
+							else
+							{
+								DeletedItemCount++;
+								DeletedByteCount += Content.FileSizes[FileIndex];
+							}
+						}
+					}
+				}
+			}
+		}
+		const std::filesystem::path& Path;
+		std::atomic<bool>&			 AbortFlag;
+		std::atomic<uint64_t>&		 DiscoveredItemCount;
+		std::atomic<uint64_t>&		 DeletedItemCount;
+		std::atomic<uint64_t>&		 DeletedByteCount;
+		std::span<const std::string> ExcludeDirectories;
+		CleanDirectoryResult&		 Result;
+		RwLock&						 ResultLock;
+	} Visitor(Path, AbortFlag, DiscoveredItemCount, DeletedItemCount, DeletedByteCount, ExcludeDirectories, Result, ResultLock);
+
+	GetDirectoryContent(Path,
+						DirectoryContentFlags::IncludeFiles | DirectoryContentFlags::Recursive | DirectoryContentFlags::IncludeFileSizes,
+						Visitor,
+						IOWorkerPool,
+						Work.PendingWork());
+
+	DirectoryContent LocalDirectoryContent;
+	GetDirectoryContent(Path, DirectoryContentFlags::IncludeDirs | DirectoryContentFlags::IncludeFiles, LocalDirectoryContent);
+	DiscoveredItemCount += LocalDirectoryContent.Directories.size();
+	std::vector<std::filesystem::path> DirectoriesToDelete;
+	DirectoriesToDelete.reserve(LocalDirectoryContent.Directories.size());
+	for (std::filesystem::path& LocalDirPath : LocalDirectoryContent.Directories)
+	{
+		bool Leave = false;
+		for (const std::string_view ExcludeDirectory : ExcludeDirectories)
+		{
+			if (LocalDirPath == (Path / ExcludeDirectory))
+			{
+				Leave = true;
+				break;
+			}
+		}
+		if (!Leave)
+		{
+			DirectoriesToDelete.emplace_back(std::move(LocalDirPath));
+			DiscoveredItemCount++;
+		}
+	}
+
+	uint64_t LastUpdateTimeMs = Timer.GetElapsedTimeMs();
+
+	if (ProgressFunc && ProgressUpdateDelayMS != 0)
+	{
+		Work.Wait(ProgressUpdateDelayMS, [&](bool IsAborted, bool IsPaused, ptrdiff_t PendingWork) {
+			ZEN_UNUSED(PendingWork);
+			LastUpdateTimeMs = Timer.GetElapsedTimeMs();
+
+			uint64_t	Deleted		 = DeletedItemCount.load();
+			uint64_t	DeletedBytes = DeletedByteCount.load();
+			uint64_t	Discovered	 = DiscoveredItemCount.load();
+			std::string Details		 = fmt::format("Found {}, Deleted {} ({})", Discovered, Deleted, NiceBytes(DeletedBytes));
+			ProgressFunc(Details, Discovered, Discovered - Deleted, IsPaused, IsAborted);
+		});
+	}
+	else
+	{
+		Work.Wait();
+	}
+
+	{
+		ZEN_TRACE_CPU("DeleteDirs");
+		for (const std::filesystem::path& DirectoryToDelete : DirectoriesToDelete)
+		{
+			if (AbortFlag)
+			{
+				break;
+			}
+			else
+			{
+				while (PauseFlag && !AbortFlag)
+				{
+					Sleep(2000);
+				}
+			}
+
+			{
+				std::error_code Ec;
+				zen::CleanDirectory(DirectoryToDelete, true, Ec);
+				if (Ec)
+				{
+					Sleep(200);
+					zen::CleanDirectory(DirectoryToDelete, true, Ec);
+				}
+
+				if (!Ec)
+				{
+					RemoveDir(Path, Ec);
+					for (size_t Retries = 0; Ec && Retries < 3; Retries++)
+					{
+						Sleep(100 + int(Retries * 50));
+						if (!IsDir(Path))
+						{
+							break;
+						}
+						Ec.clear();
+						RemoveDir(Path, Ec);
+					}
+				}
+				if (Ec)
+				{
+					RwLock::ExclusiveLockScope __(ResultLock);
+					Result.FailedRemovePaths.push_back(std::make_pair(DirectoryToDelete, Ec));
+				}
+				else
+				{
+					DeletedItemCount++;
+				}
+			}
+
+			uint64_t NowMs = Timer.GetElapsedTimeMs();
+
+			if ((NowMs - LastUpdateTimeMs) >= ProgressUpdateDelayMS)
+			{
+				LastUpdateTimeMs = NowMs;
+
+				uint64_t	Deleted		 = DeletedItemCount.load();
+				uint64_t	DeletedBytes = DeletedByteCount.load();
+				uint64_t	Discovered	 = DiscoveredItemCount.load();
+				std::string Details		 = fmt::format("Found {}, Deleted {} ({})", Discovered, Deleted, NiceBytes(DeletedBytes));
+				ProgressFunc(Details, Discovered, Discovered - Deleted, PauseFlag, AbortFlag);
+			}
+		}
+	}
+
+	return Result;
+}
+
+}  // namespace zen