Tsavorite allocator - tighten the packing of pages (#657)

* Change allocator to enqueue with the invariant that the first record of page (p+1) does not fit at the end of page (p). This allows replication to independently replay records and guarantee that they fit on the log exactly in the same way as the primary. * fixes based on comments * add another comment * add comments * fixes - we now always wrap TryAllocate with TryAllocateRetryNow. * Add Non-readcache "Insert At Tail" stress test * support 0% mutable fraction. * Fix InernalUpsert srcRecordInfo setting when found below ReadOnlyAddress Add some comments * Adjust mutable-page counts in stress test * fix typo * Enforce at least two pages of memory. * nit * update Garnet to use new allocator logic * Fix * update low memory to meet new constraint * re-enable warning * handle comments * fix bitmap tests to use at least 2 pages of memory. * fix hll tests * more testcase fixes * fix replication logic to handle micro-skips within the same page * PageAlignedShiftHeadAddress should always keep the head address, well, aligned. * update version to 1.0.25 --------- Co-authored-by: TedHartMS <[email protected]>
microsoft · Sep 20, 2024 · 810570e · 810570e
1 parent 7d866da
commit 810570e
Show file tree

Hide file tree

Showing 22 changed files with 626 additions and 273 deletions.
diff --git a/.azure/pipelines/azure-pipelines-external-release.yml b/.azure/pipelines/azure-pipelines-external-release.yml
@@ -3,7 +3,7 @@
 #      1) update the name: string below (line 6)     -- this is the version for the nuget package (e.g. 1.0.0)
 #      2) update \libs\host\GarnetServer.cs readonly string version  (~line 53)   -- NOTE - these two values need to be the same 
 ###################################### 
-name: 1.0.24
+name: 1.0.25
 trigger:
   branches:
     include:

diff --git a/libs/cluster/Server/Replication/PrimaryOps/AofTaskStore.cs b/libs/cluster/Server/Replication/PrimaryOps/AofTaskStore.cs
@@ -41,7 +41,7 @@ public AofTaskStore(ClusterProvider clusterProvider, int initialSize = 1, ILogge
                 logPageSizeMask = logPageSize - 1;
                 if (clusterProvider.serverOptions.MainMemoryReplication)
                     clusterProvider.storeWrapper.appendOnlyFile.SafeTailShiftCallback = SafeTailShiftCallback;
-                TruncateLagAddress = clusterProvider.storeWrapper.appendOnlyFile.UnsafeGetReadOnlyLagAddress() - 2 * logPageSize;
+                TruncateLagAddress = clusterProvider.storeWrapper.appendOnlyFile.UnsafeGetReadOnlyAddressLagOffset() - 2 * logPageSize;
             }
             TruncatedUntil = 0;
         }

diff --git a/libs/cluster/Server/Replication/ReplicaOps/ReplicationReplicaAofSync.cs b/libs/cluster/Server/Replication/ReplicaOps/ReplicationReplicaAofSync.cs
@@ -38,13 +38,19 @@ public unsafe void ProcessPrimaryStream(byte* record, int recordLength, long pre
 
                 if (clusterProvider.serverOptions.MainMemoryReplication)
                 {
-                    var firstRecordLength = GetFirstAofEntryLength(record);
-                    if (previousAddress > ReplicationOffset ||
-                        currentAddress >= previousAddress + firstRecordLength)
+                    // If the incoming AOF chunk fits in the space between previousAddress and currentAddress (ReplicationOffset),
+                    // an enqueue will result in an offset mismatch. So, we have to first reset the AOF to point to currentAddress.
+                    if (currentAddress > previousAddress)
                     {
-                        logger?.LogWarning("MainMemoryReplication: Skipping from {ReplicaReplicationOffset} to {currentAddress}", ReplicationOffset, currentAddress);
-                        storeWrapper.appendOnlyFile.Initialize(currentAddress, currentAddress);
-                        ReplicationOffset = currentAddress;
+                        if (
+                            (currentAddress % (1 << storeWrapper.appendOnlyFile.UnsafeGetLogPageSizeBits()) != 0) || // the skip was to a non-page-boundary
+                            (currentAddress >= previousAddress + recordLength) // the skip will not be auto-handled by the AOF enqueue
+                            )
+                        {
+                            logger?.LogWarning("MainMemoryReplication: Skipping from {ReplicaReplicationOffset} to {currentAddress}", ReplicationOffset, currentAddress);
+                            storeWrapper.appendOnlyFile.Initialize(currentAddress, currentAddress);
+                            ReplicationOffset = currentAddress;
+                        }
                     }
                 }
 
@@ -56,15 +62,6 @@ public unsafe void ProcessPrimaryStream(byte* record, int recordLength, long pre
                     throw new GarnetException($"Before ProcessPrimaryStream: Replication offset mismatch: ReplicaReplicationOffset {ReplicationOffset}, aof.TailAddress {storeWrapper.appendOnlyFile.TailAddress}", LogLevel.Warning, clientResponse: false);
                 }
 
-                // If there is a gap between the local tail and incoming currentAddress, try to skip local AOF to the next page
-                if (currentAddress >= storeWrapper.appendOnlyFile.TailAddress + recordLength
-                    && storeWrapper.appendOnlyFile.GetPage(currentAddress) == storeWrapper.appendOnlyFile.GetPage(storeWrapper.appendOnlyFile.TailAddress) + 1)
-                {
-                    logger?.LogWarning("SkipPage from {previousAddress} to {currentAddress}, tail is {tailAddress}", previousAddress, currentAddress, storeWrapper.appendOnlyFile.TailAddress);
-                    storeWrapper.appendOnlyFile.UnsafeSkipPage();
-                    logger?.LogWarning("New tail after SkipPage is {tailAddress}", storeWrapper.appendOnlyFile.TailAddress);
-                }
-
                 // Enqueue to AOF
                 _ = clusterProvider.storeWrapper.appendOnlyFile?.UnsafeEnqueueRaw(new Span<byte>(record, recordLength), noCommit: clusterProvider.serverOptions.EnableFastCommit);
 

diff --git a/libs/host/GarnetServer.cs b/libs/host/GarnetServer.cs
@@ -49,7 +49,7 @@ public class GarnetServer : IDisposable
         protected StoreWrapper storeWrapper;
 
         // IMPORTANT: Keep the version in sync with .azure\pipelines\azure-pipelines-external-release.yml line ~6.
-        readonly string version = "1.0.24";
+        readonly string version = "1.0.25";
 
         /// <summary>
         /// Resp protocol version

diff --git a/libs/server/Resp/RespServerSession.cs b/libs/server/Resp/RespServerSession.cs
@@ -432,7 +432,7 @@ private void ProcessMessages()
                         }
                         else
                         {
-                            if (CanServeSlot(cmd))
+                            if (clusterSession == null || CanServeSlot(cmd))
                                 _ = ProcessBasicCommands(cmd, ref basicGarnetApi);
                         }
                     }

diff --git a/libs/server/Resp/RespServerSessionSlotVerify.cs b/libs/server/Resp/RespServerSessionSlotVerify.cs
@@ -2,6 +2,7 @@
 // Licensed under the MIT license.
 
 using System;
+using System.Diagnostics;
 using Garnet.common;
 
 namespace Garnet.server
@@ -33,9 +34,7 @@ bool NetworkKeyArraySlotVerify(Span<ArgSlice> keys, bool readOnly, int count = -
 
         bool CanServeSlot(RespCommand cmd)
         {
-            // If cluster is disable all commands
-            if (clusterSession == null)
-                return true;
+            Debug.Assert(clusterSession != null);
 
             // Verify slot for command if it falls into data command category
             if (!cmd.IsDataCommand())