structured cache with block store

1 files changed, 53 insertions, 40 deletions

diff --git a/zenserver/cache/structuredcachestore.h b/zenserver/cache/structuredcachestore.h
index f39d01747..91ac00f35 100644
--- a/zenserver/cache/structuredcachestore.h
+++ b/zenserver/cache/structuredcachestore.h
@@ -8,6 +8,7 @@
 #include <zencore/thread.h>
 #include <zencore/uid.h>
 #include <zenstore/basicfile.h>
+#include <zenstore/blockstore.h>
 #include <zenstore/cas.h>
 #include <zenstore/caslog.h>
 #include <zenstore/gc.h>
@@ -76,37 +77,32 @@ struct DiskLocation
 {
 	inline DiskLocation() = default;
 
-	inline DiskLocation(uint64_t Offset, uint64_t ValueSize, uint32_t IndexSize, uint64_t Flags)
-	: OffsetAndFlags(CombineOffsetAndFlags(Offset, Flags))
-	, LowerSize(ValueSize & 0xFFFFffff)
-	, IndexDataSize(IndexSize)
+	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)
 	{
+		this->Location.BlockLocation = BlockStoreDiskLocation(Location, PayloadAlignment);
 	}
 
-	static const uint64_t kOffsetMask	  = 0x0000'ffFF'ffFF'ffFFull;
-	static const uint64_t kSizeMask		  = 0x00FF'0000'0000'0000ull;  // Most significant bits of value size (lower 32 bits in LowerSize)
-	static const uint64_t kFlagsMask	  = 0xff00'0000'0000'0000ull;
-	static const uint64_t kStandaloneFile = 0x8000'0000'0000'0000ull;  // Stored as a separate file
-	static const uint64_t kStructured	  = 0x4000'0000'0000'0000ull;  // Serialized as compact binary
-	static const uint64_t kTombStone	  = 0x2000'0000'0000'0000ull;  // Represents a deleted key/value
-	static const uint64_t kCompressed	  = 0x1000'0000'0000'0000ull;  // Stored in compressed buffer format
-
-	static uint64_t CombineOffsetAndFlags(uint64_t Offset, uint64_t Flags) { return Offset | Flags; }
+	inline BlockStoreLocation GetBlockLocation(uint64_t PayloadAlignment) const
+	{
+		ZEN_ASSERT(!(Flags & kStandaloneFile));
+		return Location.BlockLocation.Get(PayloadAlignment);
+	}
 
-	inline uint64_t		  Offset() const { return OffsetAndFlags & kOffsetMask; }
-	inline uint64_t		  Size() const { return LowerSize; }
-	inline uint64_t		  IsFlagSet(uint64_t Flag) const { return OffsetAndFlags & Flag; }
-	inline uint64_t		  GetFlags() const { return OffsetAndFlags & kFlagsMask; }
+	inline uint64_t		  Size() const { return (Flags & kStandaloneFile) ? Location.StandaloneSize : Location.BlockLocation.GetSize(); }
+	inline uint8_t		  IsFlagSet(uint64_t Flag) const { return Flags & Flag; }
+	inline uint8_t		  GetFlags() const { return Flags; }
 	inline ZenContentType GetContentType() const
 	{
 		ZenContentType ContentType = ZenContentType::kBinary;
 
-		if (IsFlagSet(DiskLocation::kStructured))
+		if (IsFlagSet(kStructured))
 		{
 			ContentType = ZenContentType::kCbObject;
 		}
 
-		if (IsFlagSet(DiskLocation::kCompressed))
+		if (IsFlagSet(kCompressed))
 		{
 			ContentType = ZenContentType::kCompressedBinary;
 		}
@@ -114,21 +110,29 @@ struct DiskLocation
 		return ContentType;
 	}
 
-private:
-	uint64_t OffsetAndFlags = 0;
-	uint32_t LowerSize		= 0;
-	uint32_t IndexDataSize	= 0;
+	union
+	{
+		BlockStoreDiskLocation BlockLocation;		// 10 bytes
+		uint64_t			   StandaloneSize = 0;	// 8 bytes
+	} Location;
+
+	static const uint8_t kStandaloneFile = 0x80u;  // Stored as a separate file
+	static const uint8_t kStructured	 = 0x40u;  // Serialized as compact binary
+	static const uint8_t kTombStone		 = 0x20u;  // Represents a deleted key/value
+	static const uint8_t kCompressed	 = 0x10u;  // Stored in compressed buffer format
+
+	uint8_t Flags	 = 0;
+	uint8_t Reserved = 0;
 };
 
 struct DiskIndexEntry
 {
-	IoHash		 Key;
-	DiskLocation Location;
+	IoHash		 Key;		// 20 bytes
+	DiskLocation Location;	// 12 bytes
 };
-
 #pragma pack(pop)
 
-static_assert(sizeof(DiskIndexEntry) == 36);
+static_assert(sizeof(DiskIndexEntry) == 32);
 
 /** In-memory cache storage
 
@@ -245,15 +249,18 @@ private:
 		inline uint64_t TotalSize() const { return m_TotalSize.load(std::memory_order::relaxed); }
 
 	private:
+		const uint64_t MaxBlockSize		  = 1ull << 30;
+		uint64_t	   m_PayloadAlignment = 1ull << 4;
+
 		std::string			  m_BucketName;
 		std::filesystem::path m_BucketDir;
+		std::filesystem::path m_BlocksBasePath;
 		Oid					  m_BucketId;
 		bool				  m_IsOk				 = false;
 		uint64_t			  m_LargeObjectThreshold = 64 * 1024;
 
 		// These files are used to manage storage of small objects for this bucket
 
-		BasicFile					m_SobsFile;
 		TCasLogFile<DiskIndexEntry> m_SlogFile;
 
 		struct IndexEntry
@@ -277,20 +284,26 @@ private:
 
 		using IndexMap = tsl::robin_map<IoHash, IndexEntry, IoHash::Hasher>;
 
-		RwLock				 m_IndexLock;
-		IndexMap			 m_Index;
-		uint64_t			 m_SobsCursor = 0;
+		RwLock											  m_IndexLock;
+		IndexMap										  m_Index;
+		std::unordered_map<uint32_t, Ref<BlockStoreFile>> m_ChunkBlocks;
+
+		RwLock				m_InsertLock;  // used to serialize inserts
+		Ref<BlockStoreFile> m_WriteBlock;
+		std::uint64_t		m_CurrentInsertOffset = 0;
+
+		std::atomic_uint32_t m_WriteBlockIndex{};
 		std::atomic_uint64_t m_TotalSize{};
 
-		void BuildPath(PathBuilderBase& Path, const IoHash& HashKey);
-		void PutStandaloneCacheValue(const IoHash& HashKey, const ZenCacheValue& Value);
-		bool GetStandaloneCacheValue(const DiskLocation& Loc, const IoHash& HashKey, ZenCacheValue& OutValue);
-		void DeleteStandaloneCacheValue(const DiskLocation&			 Loc,
-										const IoHash&				 HashKey,
-										const std::filesystem::path& Path,
-										std::error_code&			 Ec);
-		bool GetInlineCacheValue(const DiskLocation& Loc, ZenCacheValue& OutValue);
-		void OpenLog(const std::filesystem::path& BucketDir, const bool IsNew);
+		void	 BuildPath(PathBuilderBase& Path, const IoHash& HashKey);
+		void	 PutStandaloneCacheValue(const IoHash& HashKey, const ZenCacheValue& Value);
+		bool	 GetStandaloneCacheValue(const DiskLocation& Loc, const IoHash& HashKey, ZenCacheValue& OutValue);
+		bool	 GetInlineCacheValue(const DiskLocation& Loc, ZenCacheValue& OutValue);
+		void	 MakeIndexSnapshot();
+		uint64_t ReadIndexFile();
+		uint64_t ReadLog(uint64_t LogPosition);
+		uint64_t MigrateLegacyData(bool CleanSource);
+		void	 OpenLog(const std::filesystem::path& BucketDir, const bool IsNew);
 
 		// These locks are here to avoid contention on file creation, therefore it's sufficient
 		// that we take the same lock for the same hash