Tsavorite kernel

libs/cluster/Server/ClusterManagerSlotState.cs

@@ -12,12 +12,7 @@
 
 namespace Garnet.cluster
 {
-    using BasicGarnetApi = GarnetApi<BasicContext<SpanByte, SpanByte, SpanByte, SpanByteAndMemory, long, MainSessionFunctions,
-            /* MainStoreFunctions */ StoreFunctions<SpanByte, SpanByte, SpanByteComparer, SpanByteRecordDisposer>,
-            SpanByteAllocator<StoreFunctions<SpanByte, SpanByte, SpanByteComparer, SpanByteRecordDisposer>>>,
-        BasicContext<byte[], IGarnetObject, ObjectInput, GarnetObjectStoreOutput, long, ObjectSessionFunctions,
-            /* ObjectStoreFunctions */ StoreFunctions<byte[], IGarnetObject, ByteArrayKeyComparer, DefaultRecordDisposer<byte[], IGarnetObject>>,
-            GenericAllocator<byte[], IGarnetObject, StoreFunctions<byte[], IGarnetObject, ByteArrayKeyComparer, DefaultRecordDisposer<byte[], IGarnetObject>>>>>;
+    using BasicGarnetApi = GarnetApi<TransientKeyLocker, GarnetSafeEpochGuard>;
 
     /// <summary>
     /// Cluster manager

libs/cluster/Server/ClusterProvider.cs

@@ -15,12 +15,7 @@
 
 namespace Garnet.cluster
 {
-    using BasicGarnetApi = GarnetApi<BasicContext<SpanByte, SpanByte, SpanByte, SpanByteAndMemory, long, MainSessionFunctions,
-            /* MainStoreFunctions */ StoreFunctions<SpanByte, SpanByte, SpanByteComparer, SpanByteRecordDisposer>,
-            SpanByteAllocator<StoreFunctions<SpanByte, SpanByte, SpanByteComparer, SpanByteRecordDisposer>>>,
-        BasicContext<byte[], IGarnetObject, ObjectInput, GarnetObjectStoreOutput, long, ObjectSessionFunctions,
-            /* ObjectStoreFunctions */ StoreFunctions<byte[], IGarnetObject, ByteArrayKeyComparer, DefaultRecordDisposer<byte[], IGarnetObject>>,
-            GenericAllocator<byte[], IGarnetObject, StoreFunctions<byte[], IGarnetObject, ByteArrayKeyComparer, DefaultRecordDisposer<byte[], IGarnetObject>>>>>;
+    using BasicGarnetApi = GarnetApi<TransientKeyLocker, GarnetSafeEpochGuard>;
 
     /// <summary>
     /// Cluster provider
@@ -142,8 +137,8 @@ public void FlushConfig()
         /// <inheritdoc />
         public void FlushDB(bool unsafeTruncateLog = false)
         {
-            storeWrapper.store.Log.ShiftBeginAddress(storeWrapper.store.Log.TailAddress, truncateLog: unsafeTruncateLog);
-            storeWrapper.objectStore?.Log.ShiftBeginAddress(storeWrapper.objectStore.Log.TailAddress, truncateLog: unsafeTruncateLog);
+            storeWrapper.Store.Log.ShiftBeginAddress(storeWrapper.Store.Log.TailAddress, truncateLog: unsafeTruncateLog);
+            storeWrapper.ObjectStore?.Log.ShiftBeginAddress(storeWrapper.ObjectStore.Log.TailAddress, truncateLog: unsafeTruncateLog);
         }
 
         /// <inheritdoc />
@@ -153,7 +148,7 @@ public void SafeTruncateAOF(StoreType storeType, bool full, long CheckpointCover
 
             if (storeType is StoreType.Main or StoreType.All)
             {
-                entry.storeVersion = storeWrapper.store.CurrentVersion;
+                entry.storeVersion = storeWrapper.Store.CurrentVersion;
                 entry.storeHlogToken = storeCheckpointToken;
                 entry.storeIndexToken = storeCheckpointToken;
                 entry.storeCheckpointCoveredAofAddress = CheckpointCoveredAofAddress;
@@ -162,7 +157,7 @@ public void SafeTruncateAOF(StoreType storeType, bool full, long CheckpointCover
 
             if (storeType is StoreType.Object or StoreType.All)
             {
-                entry.objectStoreVersion = serverOptions.DisableObjects ? -1 : storeWrapper.objectStore.CurrentVersion;
+                entry.objectStoreVersion = serverOptions.DisableObjects ? -1 : storeWrapper.ObjectStore.CurrentVersion;
                 entry.objectStoreHlogToken = serverOptions.DisableObjects ? default : objectStoreCheckpointToken;
                 entry.objectStoreIndexToken = serverOptions.DisableObjects ? default : objectStoreCheckpointToken;
                 entry.objectCheckpointCoveredAofAddress = CheckpointCoveredAofAddress;
@@ -290,8 +285,8 @@ internal ReplicationLogCheckpointManager GetReplicationLogCheckpointManager(Stor
             Debug.Assert(serverOptions.EnableCluster);
             return storeType switch
             {
-                StoreType.Main => (ReplicationLogCheckpointManager)storeWrapper.store.CheckpointManager,
-                StoreType.Object => (ReplicationLogCheckpointManager)storeWrapper.objectStore?.CheckpointManager,
+                StoreType.Main => (ReplicationLogCheckpointManager)storeWrapper.Store.CheckpointManager,
+                StoreType.Object => (ReplicationLogCheckpointManager)storeWrapper.ObjectStore?.CheckpointManager,
                 _ => throw new Exception($"GetCkptManager: unexpected state {storeType}")
             };
         }

libs/cluster/Server/Migration/MigrateSessionKeys.cs

@@ -193,14 +193,14 @@ public bool MigrateKeys()
                     return false;
 
                 // Migrate main store keys
-                _gcs.InitMigrateBuffer(clusterProvider.storeWrapper.loggingFrequncy);
+                _gcs.InitMigrateBuffer(clusterProvider.storeWrapper.loggingFrequency);
                 if (!MigrateKeysFromMainStore())
                     return false;
 
                 // Migrate object store keys
                 if (!clusterProvider.serverOptions.DisableObjects)
                 {
-                    _gcs.InitMigrateBuffer(clusterProvider.storeWrapper.loggingFrequncy);
+                    _gcs.InitMigrateBuffer(clusterProvider.storeWrapper.loggingFrequency);
                     if (!MigrateKeysFromObjectStore())
                         return false;
                 }

libs/cluster/Server/Migration/MigrateSessionSlots.cs

@@ -15,7 +15,7 @@ internal sealed unsafe partial class MigrateSession : IDisposable
         public bool MigrateSlotsDriver()
         {
             logger?.LogTrace("Initializing MainStore Iterator");
-            var storeTailAddress = clusterProvider.storeWrapper.store.Log.TailAddress;
+            var storeTailAddress = clusterProvider.storeWrapper.Store.Log.TailAddress;
             var bufferSize = 1 << clusterProvider.serverOptions.PageSizeBits();
             MigrationKeyIterationFunctions.MainStoreGetKeysInSlots mainStoreGetKeysInSlots = new(this, _sslots, bufferSize: bufferSize);
 
@@ -55,7 +55,7 @@ public bool MigrateSlotsDriver()
             if (!clusterProvider.serverOptions.DisableObjects)
             {
                 logger?.LogTrace("Initializing ObjectStore Iterator");
-                var objectStoreTailAddress = clusterProvider.storeWrapper.objectStore.Log.TailAddress;
+                var objectStoreTailAddress = clusterProvider.storeWrapper.ObjectStore.Log.TailAddress;
                 var objectBufferSize = 1 << clusterProvider.serverOptions.ObjectStorePageSizeBits();
                 MigrationKeyIterationFunctions.ObjectStoreGetKeysInSlots objectStoreGetKeysInSlots = new(this, _sslots, bufferSize: objectBufferSize);
 

libs/cluster/Server/Replication/CheckpointStore.cs

@@ -272,20 +272,20 @@ public CheckpointEntry GetLatestCheckpointEntryFromDisk()
         {
             Guid objectStoreHLogToken = default;
             Guid objectStoreIndexToken = default;
-            storeWrapper.store.GetLatestCheckpointTokens(out var storeHLogToken, out var storeIndexToken);
-            storeWrapper.objectStore?.GetLatestCheckpointTokens(out objectStoreHLogToken, out objectStoreIndexToken);
+            storeWrapper.Store.GetLatestCheckpointTokens(out var storeHLogToken, out var storeIndexToken);
+            storeWrapper.ObjectStore?.GetLatestCheckpointTokens(out objectStoreHLogToken, out objectStoreIndexToken);
             var (storeCheckpointCoveredAofAddress, storePrimaryReplId) = GetCheckpointCookieMetadata(StoreType.Main, storeHLogToken);
             var (objectCheckpointCoveredAofAddress, objectStorePrimaryReplId) = objectStoreHLogToken == default ? (long.MaxValue, null) : GetCheckpointCookieMetadata(StoreType.Object, objectStoreHLogToken);
 
             CheckpointEntry entry = new()
             {
-                storeVersion = storeHLogToken == default ? -1 : storeWrapper.store.GetLatestCheckpointVersion(),
+                storeVersion = storeHLogToken == default ? -1 : storeWrapper.Store.GetLatestCheckpointVersion(),
                 storeHlogToken = storeHLogToken,
                 storeIndexToken = storeIndexToken,
                 storeCheckpointCoveredAofAddress = storeCheckpointCoveredAofAddress,
                 storePrimaryReplId = storePrimaryReplId,
 
-                objectStoreVersion = objectStoreHLogToken == default ? -1 : storeWrapper.objectStore.GetLatestCheckpointVersion(),
+                objectStoreVersion = objectStoreHLogToken == default ? -1 : storeWrapper.ObjectStore.GetLatestCheckpointVersion(),
                 objectStoreHlogToken = objectStoreHLogToken,
                 objectStoreIndexToken = objectStoreIndexToken,
                 objectCheckpointCoveredAofAddress = objectCheckpointCoveredAofAddress,

libs/cluster/Server/Replication/ReplicationCheckpointManagement.cs

@@ -31,8 +31,8 @@ public bool TryAcquireSettledMetadataForMainStore(CheckpointEntry entry, out Log
             index_size = -1;
             try
             {
-                hlog_size = storeWrapper.store.GetLogFileSize(entry.storeHlogToken);
-                index_size = storeWrapper.store.GetIndexFileSize(entry.storeIndexToken);
+                hlog_size = storeWrapper.Store.GetLogFileSize(entry.storeHlogToken);
+                index_size = storeWrapper.Store.GetIndexFileSize(entry.storeIndexToken);
                 return true;
             }
             catch
@@ -54,8 +54,8 @@ public bool TryAcquireSettledMetadataForObjectStore(CheckpointEntry entry, out L
             index_size = -1;
             try
             {
-                hlog_size = storeWrapper.objectStore.GetLogFileSize(entry.objectStoreHlogToken);
-                index_size = storeWrapper.objectStore.GetIndexFileSize(entry.objectStoreIndexToken);
+                hlog_size = storeWrapper.ObjectStore.GetLogFileSize(entry.objectStoreHlogToken);
+                index_size = storeWrapper.ObjectStore.GetIndexFileSize(entry.objectStoreIndexToken);
                 return true;
             }
             catch

libs/cluster/Server/Replication/ReplicationManager.cs

@@ -66,8 +66,8 @@ public ReplicationLogCheckpointManager GetCkptManager(StoreType storeType)
         {
             return storeType switch
             {
-                StoreType.Main => (ReplicationLogCheckpointManager)storeWrapper.store.CheckpointManager,
-                StoreType.Object => (ReplicationLogCheckpointManager)storeWrapper.objectStore?.CheckpointManager,
+                StoreType.Main => (ReplicationLogCheckpointManager)storeWrapper.Store.CheckpointManager,
+                StoreType.Object => (ReplicationLogCheckpointManager)storeWrapper.ObjectStore?.CheckpointManager,
                 _ => throw new Exception($"GetCkptManager: unexpected state {storeType}")
             };
         }
@@ -103,7 +103,7 @@ public ReplicationManager(ClusterProvider clusterProvider, ILogger logger = null
 
             // Set the appendOnlyFile field for all stores
             clusterProvider.GetReplicationLogCheckpointManager(StoreType.Main).checkpointVersionShift = CheckpointVersionShift;
-            if (storeWrapper.objectStore != null)
+            if (storeWrapper.ObjectStore != null)
                 clusterProvider.GetReplicationLogCheckpointManager(StoreType.Object).checkpointVersionShift = CheckpointVersionShift;
 
             // If this node starts as replica, it cannot serve requests until it is connected to primary

libs/cluster/Session/ClusterSession.cs

@@ -12,12 +12,7 @@
 
 namespace Garnet.cluster
 {
-    using BasicGarnetApi = GarnetApi<BasicContext<SpanByte, SpanByte, SpanByte, SpanByteAndMemory, long, MainSessionFunctions,
-            /* MainStoreFunctions */ StoreFunctions<SpanByte, SpanByte, SpanByteComparer, SpanByteRecordDisposer>,
-            SpanByteAllocator<StoreFunctions<SpanByte, SpanByte, SpanByteComparer, SpanByteRecordDisposer>>>,
-        BasicContext<byte[], IGarnetObject, ObjectInput, GarnetObjectStoreOutput, long, ObjectSessionFunctions,
-            /* ObjectStoreFunctions */ StoreFunctions<byte[], IGarnetObject, ByteArrayKeyComparer, DefaultRecordDisposer<byte[], IGarnetObject>>,
-            GenericAllocator<byte[], IGarnetObject, StoreFunctions<byte[], IGarnetObject, ByteArrayKeyComparer, DefaultRecordDisposer<byte[], IGarnetObject>>>>>;
+    using BasicGarnetApi = GarnetApi<TransientKeyLocker, GarnetSafeEpochGuard>;
 
     internal sealed unsafe partial class ClusterSession : IClusterSession
     {
@@ -50,7 +45,8 @@ internal sealed unsafe partial class ClusterSession : IClusterSession
         public void SetReadOnlySession() => readWriteSession = false;
         public void SetReadWriteSession() => readWriteSession = true;
 
-        public ClusterSession(ClusterProvider clusterProvider, TransactionManager txnManager, IGarnetAuthenticator authenticator, User user, GarnetSessionMetrics sessionMetrics, BasicGarnetApi basicGarnetApi, INetworkSender networkSender, ILogger logger = null)
+        public ClusterSession(ClusterProvider clusterProvider, TransactionManager txnManager, IGarnetAuthenticator authenticator, User user, GarnetSessionMetrics sessionMetrics, 
+                BasicGarnetApi basicGarnetApi, INetworkSender networkSender, ILogger logger = null)
         {
             this.clusterProvider = clusterProvider;
             this.authenticator = authenticator;

libs/cluster/Session/RespClusterMigrateCommands.cs

@@ -26,7 +26,7 @@ private void TrackImportProgress(int keyCount, bool isMainStore, bool completed
         {
             totalKeyCount += keyCount;
             var duration = TimeSpan.FromTicks(Stopwatch.GetTimestamp() - lastLog);
-            if (completed || lastLog == 0 || duration >= clusterProvider.storeWrapper.loggingFrequncy)
+            if (completed || lastLog == 0 || duration >= clusterProvider.storeWrapper.loggingFrequency)
             {
                 logger?.LogTrace("[{op}]: isMainStore:({storeType}) totalKeyCount:({totalKeyCount})", completed ? "COMPLETED" : "IMPORTING", isMainStore, totalKeyCount.ToString("N0"));
                 lastLog = Stopwatch.GetTimestamp();