Merge branch 'main' of https://github.com/EpicGames/zen

1 files changed, 106 insertions, 57 deletions

diff --git a/src/zenserver/cache/cachedisklayer.h b/src/zenserver/cache/cachedisklayer.h
index d46d629e4..6997a12e4 100644
--- a/src/zenserver/cache/cachedisklayer.h
+++ b/src/zenserver/cache/cachedisklayer.h
@@ -29,14 +29,25 @@ struct DiskLocation
 
 	inline DiskLocation(uint64_t ValueSize, uint8_t Flags) : Flags(Flags | kStandaloneFile) { Location.StandaloneSize = ValueSize; }
 
-	inline DiskLocation(const BlockStoreLocation& Location, uint64_t PayloadAlignment, uint8_t Flags) : Flags(Flags & ~kStandaloneFile)
+	inline DiskLocation(const BlockStoreLocation& Location, uint32_t PayloadAlignment, uint8_t Flags) : Flags(Flags & ~kStandaloneFile)
 	{
 		this->Location.BlockLocation = BlockStoreDiskLocation(Location, PayloadAlignment);
 	}
 
-	inline bool operator!=(const DiskLocation& Rhs) const { return memcmp(&Location, &Rhs.Location, sizeof(Location)) != 0; }
+	inline bool operator!=(const DiskLocation& Rhs) const
+	{
+		if (Flags != Rhs.Flags)
+		{
+			return true;
+		}
+		if (Flags & kStandaloneFile)
+		{
+			return Location.StandaloneSize != Rhs.Location.StandaloneSize;
+		}
+		return Location.BlockLocation != Rhs.Location.BlockLocation;
+	}
 
-	inline BlockStoreLocation GetBlockLocation(uint64_t PayloadAlignment) const
+	inline BlockStoreLocation GetBlockLocation(uint32_t PayloadAlignment) const
 	{
 		ZEN_ASSERT(!(Flags & kStandaloneFile));
 		return Location.BlockLocation.Get(PayloadAlignment);
@@ -95,7 +106,7 @@ public:
 	struct BucketConfiguration
 	{
 		uint64_t MaxBlockSize			= 1ull << 30;
-		uint64_t PayloadAlignment		= 1ull << 4;
+		uint32_t PayloadAlignment		= 1u << 4;
 		uint64_t MemCacheSizeThreshold	= 1 * 1024;
 		uint64_t LargeObjectThreshold	= 128 * 1024;
 		bool	 EnableReferenceCaching = false;
@@ -178,7 +189,6 @@ public:
 	void SetAccessTime(std::string_view Bucket, const IoHash& HashKey, GcClock::TimePoint Time);
 #endif	// ZEN_WITH_TESTS
 
-private:
 	/** A cache bucket manages a single directory containing
 		metadata and data for that bucket
 	  */
@@ -187,15 +197,15 @@ private:
 		CacheBucket(GcManager& Gc, std::atomic_uint64_t& OuterCacheMemoryUsage, std::string BucketName, const BucketConfiguration& Config);
 		~CacheBucket();
 
-		bool OpenOrCreate(std::filesystem::path BucketDir, bool AllowCreate = true);
-		bool Get(const IoHash& HashKey, ZenCacheValue& OutValue);
-		void Put(const IoHash& HashKey, const ZenCacheValue& Value, std::span<IoHash> References);
-		void MemCacheTrim(GcClock::TimePoint ExpireTime);
-		bool Drop();
-		void Flush();
-		void ScrubStorage(ScrubContext& Ctx);
-		void GatherReferences(GcContext& GcCtx);
-		void CollectGarbage(GcContext& GcCtx);
+		bool	 OpenOrCreate(std::filesystem::path BucketDir, bool AllowCreate = true);
+		bool	 Get(const IoHash& HashKey, ZenCacheValue& OutValue);
+		void	 Put(const IoHash& HashKey, const ZenCacheValue& Value, std::span<IoHash> References);
+		uint64_t MemCacheTrim(GcClock::TimePoint ExpireTime);
+		bool	 Drop();
+		void	 Flush();
+		void	 ScrubStorage(ScrubContext& Ctx);
+		void	 GatherReferences(GcContext& GcCtx);
+		void	 CollectGarbage(GcContext& GcCtx);
 
 		inline GcStorageSize StorageSize() const
 		{
@@ -205,30 +215,15 @@ private:
 		uint64_t	EntryCount() const;
 		BucketStats Stats();
 
-		CacheValueDetails::BucketDetails GetValueDetails(const std::string_view ValueFilter) const;
+		CacheValueDetails::BucketDetails GetValueDetails(RwLock::SharedLockScope& IndexLock, const std::string_view ValueFilter) const;
 		void EnumerateBucketContents(std::function<void(const IoHash& Key, const CacheValueDetails::ValueDetails& Details)>& Fn) const;
 
-		void GetUsageByAccess(GcClock::TimePoint TickStart, GcClock::Duration SectionLength, std::vector<uint64_t>& InOutUsageSlots);
+		void GetUsageByAccess(GcClock::TimePoint Now, GcClock::Duration MaxAge, std::vector<uint64_t>& InOutUsageSlots);
 #if ZEN_WITH_TESTS
 		void SetAccessTime(const IoHash& HashKey, GcClock::TimePoint Time);
 #endif	// ZEN_WITH_TESTS
 
 	private:
-		GcManager&			  m_Gc;
-		std::atomic_uint64_t& m_OuterCacheMemoryUsage;
-		std::string			  m_BucketName;
-		std::filesystem::path m_BucketDir;
-		std::filesystem::path m_BlocksBasePath;
-		BucketConfiguration	  m_Configuration;
-		BlockStore			  m_BlockStore;
-		Oid					  m_BucketId;
-		std::atomic_bool	  m_IsFlushing{};
-
-		// These files are used to manage storage of small objects for this bucket
-
-		TCasLogFile<DiskIndexEntry> m_SlogFile;
-		uint64_t					m_LogFlushPosition = 0;
-
 #pragma pack(push)
 #pragma pack(1)
 		struct MetaDataIndex
@@ -291,6 +286,11 @@ private:
 
 			operator bool() const { return RawSize != 0 || RawHash != IoHash::Zero; };
 		};
+		struct MemCacheData
+		{
+			IoBuffer	 Payload;
+			PayloadIndex OwnerIndex;
+		};
 #pragma pack(pop)
 		static_assert(sizeof(BucketPayload) == 20u);
 		static_assert(sizeof(BucketMetaData) == 28u);
@@ -298,6 +298,21 @@ private:
 
 		using IndexMap = tsl::robin_map<IoHash, PayloadIndex, IoHash::Hasher>;
 
+		GcManager&			  m_Gc;
+		std::atomic_uint64_t& m_OuterCacheMemoryUsage;
+		std::string			  m_BucketName;
+		std::filesystem::path m_BucketDir;
+		std::filesystem::path m_BlocksBasePath;
+		BucketConfiguration	  m_Configuration;
+		BlockStore			  m_BlockStore;
+		Oid					  m_BucketId;
+		std::atomic_bool	  m_IsFlushing{true};  // Don't allow flush until we are properly initialized
+
+		// These files are used to manage storage of small objects for this bucket
+
+		TCasLogFile<DiskIndexEntry> m_SlogFile;
+		uint64_t					m_LogFlushPosition = 0;
+
 		std::atomic<uint64_t> m_DiskHitCount;
 		std::atomic<uint64_t> m_DiskMissCount;
 		std::atomic<uint64_t> m_DiskWriteCount;
@@ -313,17 +328,18 @@ private:
 		std::vector<BucketPayload>	m_Payloads;
 		std::vector<BucketMetaData> m_MetaDatas;
 		std::vector<MetaDataIndex>	m_FreeMetaDatas;
-		std::vector<IoBuffer>		m_MemCachedPayloads;
+		std::vector<MemCacheData>	m_MemCachedPayloads;
 		std::vector<MemCachedIndex> m_FreeMemCachedPayloads;
 		std::vector<ReferenceIndex> m_FirstReferenceIndex;
 		std::vector<IoHash>			m_ReferenceHashes;
 		std::vector<ReferenceIndex> m_NextReferenceHashesIndexes;
+		std::unique_ptr<HashSet>	m_UpdatedKeys;
 		size_t						m_ReferenceCount = 0;
 		std::atomic_uint64_t		m_StandaloneSize{};
 		std::atomic_uint64_t		m_MemCachedSize{};
 
 		virtual std::string						 GetGcName(GcCtx& Ctx) override;
-		virtual void							 RemoveExpiredData(GcCtx& Ctx, GcReferencerStats& Stats) override;
+		virtual GcStoreCompactor*				 RemoveExpiredData(GcCtx& Ctx, GcStats& Stats) override;
 		virtual std::vector<GcReferenceChecker*> CreateReferenceCheckers(GcCtx& Ctx) override;
 
 		void	 BuildPath(PathBuilderBase& Path, const IoHash& HashKey) const;
@@ -331,19 +347,9 @@ private:
 		IoBuffer GetStandaloneCacheValue(ZenContentType ContentType, const IoHash& HashKey) const;
 		void	 PutInlineCacheValue(const IoHash& HashKey, const ZenCacheValue& Value, std::span<IoHash> References);
 		IoBuffer GetInlineCacheValue(const DiskLocation& Loc) const;
-		void	 MakeIndexSnapshot(const std::function<uint64_t()>& ClaimDiskReserveFunc = []() { return 0; });
-		uint64_t ReadIndexFile(const std::filesystem::path& IndexPath, uint32_t& OutVersion);
-		uint64_t ReadLog(const std::filesystem::path& LogPath, uint64_t LogPosition);
-		void	 OpenLog(const bool IsNew);
-		CbObject MakeManifest(IndexMap&&						 Index,
-							  std::vector<AccessTime>&&			 AccessTimes,
-							  const std::vector<BucketPayload>&	 Payloads,
-							  const std::vector<BucketMetaData>& MetaDatas);
-		void	 SaveManifest(
-				CbObject&&						 Manifest,
-				const std::function<uint64_t()>& ClaimDiskReserveFunc = []() { return 0; });
-		CacheValueDetails::ValueDetails GetValueDetails(const IoHash& Key, PayloadIndex Index) const;
-		void							CompactReferences(RwLock::ExclusiveLockScope&);
+		CacheValueDetails::ValueDetails GetValueDetails(RwLock::SharedLockScope&, const IoHash& Key, PayloadIndex Index) const;
+
+		void		CompactReferences(RwLock::ExclusiveLockScope&);
 		void		SetReferences(RwLock::ExclusiveLockScope&, ReferenceIndex& FirstReferenceIndex, std::span<IoHash> References);
 		void		RemoveReferences(RwLock::ExclusiveLockScope&, ReferenceIndex& FirstReferenceIndex);
 		inline bool GetReferences(RwLock::SharedLockScope&, ReferenceIndex FirstReferenceIndex, std::vector<IoHash>& OutReferences) const
@@ -357,16 +363,39 @@ private:
 		ReferenceIndex AllocateReferenceEntry(RwLock::ExclusiveLockScope&, const IoHash& Key);
 		bool		   LockedGetReferences(ReferenceIndex FirstReferenceIndex, std::vector<IoHash>& OutReferences) const;
 		void		   ClearReferenceCache();
-		void		   SetMetaData(BucketPayload& Payload, const ZenCacheDiskLayer::CacheBucket::BucketMetaData& MetaData);
-		void		   RemoveMetaData(BucketPayload& Payload);
-		BucketMetaData GetMetaData(const BucketPayload& Payload) const;
-		void		   SetMemCachedData(BucketPayload& Payload, IoBuffer& MemCachedData);
-		size_t		   RemoveMemCachedData(BucketPayload& Payload);
 
-		void CompactState(std::vector<BucketPayload>&  Payloads,
+		void		   SetMetaData(RwLock::ExclusiveLockScope&,
+								   BucketPayload&										 Payload,
+								   const ZenCacheDiskLayer::CacheBucket::BucketMetaData& MetaData);
+		void		   RemoveMetaData(RwLock::ExclusiveLockScope&, BucketPayload& Payload);
+		BucketMetaData GetMetaData(RwLock::SharedLockScope&, const BucketPayload& Payload) const;
+		void		   SetMemCachedData(RwLock::ExclusiveLockScope&, PayloadIndex PayloadIndex, IoBuffer& MemCachedData);
+		size_t		   RemoveMemCachedData(RwLock::ExclusiveLockScope&, BucketPayload& Payload);
+
+		void	 InitializeIndexFromDisk(RwLock::ExclusiveLockScope&, bool IsNew);
+		uint64_t ReadIndexFile(RwLock::ExclusiveLockScope&, const std::filesystem::path& IndexPath, uint32_t& OutVersion);
+		uint64_t ReadLog(RwLock::ExclusiveLockScope&, const std::filesystem::path& LogPath, uint64_t LogPosition);
+
+		void SaveSnapshot(const std::function<uint64_t()>& ClaimDiskReserveFunc = []() { return 0; });
+		void WriteIndexSnapshot(
+			RwLock::ExclusiveLockScope&,
+			const std::function<uint64_t()>& ClaimDiskReserveFunc = []() { return 0; })
+		{
+			WriteIndexSnapshotLocked(ClaimDiskReserveFunc);
+		}
+		void WriteIndexSnapshot(
+			RwLock::SharedLockScope&,
+			const std::function<uint64_t()>& ClaimDiskReserveFunc = []() { return 0; })
+		{
+			WriteIndexSnapshotLocked(ClaimDiskReserveFunc);
+		}
+		void WriteIndexSnapshotLocked(const std::function<uint64_t()>& ClaimDiskReserveFunc = []() { return 0; });
+
+		void CompactState(RwLock::ExclusiveLockScope&,
+						  std::vector<BucketPayload>&  Payloads,
 						  std::vector<AccessTime>&	   AccessTimes,
 						  std::vector<BucketMetaData>& MetaDatas,
-						  std::vector<IoBuffer>&	   MemCachedPayloads,
+						  std::vector<MemCacheData>&   MemCachedPayloads,
 						  std::vector<ReferenceIndex>& FirstReferenceIndex,
 						  IndexMap&					   Index,
 						  RwLock::ExclusiveLockScope&  IndexLock);
@@ -381,6 +410,10 @@ private:
 			m_MemCachedSize.fetch_sub(ValueSize, std::memory_order::relaxed);
 			m_OuterCacheMemoryUsage.fetch_sub(ValueSize, std::memory_order::relaxed);
 		}
+		static inline uint64_t EstimateMemCachePayloadMemory(uint64_t PayloadSize)
+		{
+			return sizeof(MemCacheData) + sizeof(IoBufferCore) + RoundUp(PayloadSize, 8u);
+		}
 
 		// These locks are here to avoid contention on file creation, therefore it's sufficient
 		// that we take the same lock for the same hash
@@ -392,9 +425,13 @@ private:
 		inline RwLock& LockForHash(const IoHash& Hash) const { return m_ShardedLocks[Hash.Hash[19]]; }
 
 		friend class DiskBucketReferenceChecker;
+		friend class DiskBucketStoreCompactor;
+		friend class BucketManifestSerializer;
 	};
 
-	inline void TryMemCacheTrim()
+private:
+	CacheBucket* GetOrCreateBucket(std::string_view InBucket);
+	inline void	 TryMemCacheTrim()
 	{
 		if (m_Configuration.MemCacheTargetFootprintBytes == 0)
 		{
@@ -408,10 +445,21 @@ private:
 		{
 			return;
 		}
+		if (m_IsMemCacheTrimming)
+		{
+			return;
+		}
+
+		const GcClock::Tick NowTick = GcClock::TickCount();
+		if (NowTick < m_NextAllowedTrimTick)
+		{
+			return;
+		}
+
 		MemCacheTrim();
 	}
-	void MemCacheTrim();
-	void MemCacheTrim(std::vector<CacheBucket*>& Buckets, GcClock::TimePoint ExpireTime);
+	void	 MemCacheTrim();
+	uint64_t MemCacheTrim(std::vector<CacheBucket*>& Buckets, GcClock::TimePoint ExpireTime);
 
 	GcManager&													  m_Gc;
 	JobQueue&													  m_JobQueue;
@@ -419,7 +467,7 @@ private:
 	Configuration												  m_Configuration;
 	std::atomic_uint64_t										  m_TotalMemCachedSize{};
 	std::atomic_bool											  m_IsMemCacheTrimming = false;
-	std::atomic<GcClock::Tick>									  m_LastTickMemCacheTrim;
+	std::atomic<GcClock::Tick>									  m_NextAllowedTrimTick;
 	mutable RwLock												  m_Lock;
 	std::unordered_map<std::string, std::unique_ptr<CacheBucket>> m_Buckets;
 	std::vector<std::unique_ptr<CacheBucket>>					  m_DroppedBuckets;
@@ -427,6 +475,7 @@ private:
 	ZenCacheDiskLayer(const ZenCacheDiskLayer&) = delete;
 	ZenCacheDiskLayer& operator=(const ZenCacheDiskLayer&) = delete;
 
+	friend class DiskBucketStoreCompactor;
 	friend class DiskBucketReferenceChecker;
 };