Spanify TsavoriteLogRecoveryInfo deserialization (#145)

microsoft · Mar 26, 2024 · dd602d4 · dd602d4
1 parent 2aea8fb
commit dd602d4
Show file tree

Hide file tree

Showing 4 changed files with 45 additions and 47 deletions.
diff --git a/libs/cluster/Server/Replication/ReplicaOps/ReplicationReplicaAofSync.cs b/libs/cluster/Server/Replication/ReplicaOps/ReplicationReplicaAofSync.cs
@@ -79,7 +79,7 @@ public unsafe void ProcessPrimaryStream(byte* record, int recordLength, long pre
                             throw new Exception("Received FastCommit request at replica AOF processor, but FastCommit is not enabled");
                         }
                         TsavoriteLogRecoveryInfo info = new();
-                        info.Initialize(new BinaryReader(new UnmanagedMemoryStream(ptr + entryLength, -payloadLength)));
+                        info.Initialize(new ReadOnlySpan<byte>(ptr + entryLength, -payloadLength));
                         storeWrapper.appendOnlyFile?.UnsafeCommitMetadataOnly(info);
                         entryLength += storeWrapper.appendOnlyFile.UnsafeAlign(-payloadLength);
                     }

diff --git a/libs/storage/Tsavorite/cs/src/core/TsavoriteLog/TsavoriteLog.cs b/libs/storage/Tsavorite/cs/src/core/TsavoriteLog/TsavoriteLog.cs
@@ -2254,10 +2254,7 @@ private bool LoadCommitMetadata(long commitNum, out TsavoriteLogRecoveryInfo inf
             }
 
             info = new TsavoriteLogRecoveryInfo();
-            using (BinaryReader r = new(new MemoryStream(commitInfo)))
-            {
-                info.Initialize(r);
-            }
+            info.Initialize(commitInfo);
 
             if (info.CommitNum == -1)
                 info.CommitNum = commitNum;

diff --git a/libs/storage/Tsavorite/cs/src/core/TsavoriteLog/TsavoriteLogIterator.cs b/libs/storage/Tsavorite/cs/src/core/TsavoriteLog/TsavoriteLogIterator.cs
@@ -282,7 +282,7 @@ public unsafe bool GetNext(out byte[] entry, out int entryLength, out long curre
                 if (isCommitRecord)
                 {
                     TsavoriteLogRecoveryInfo info = new();
-                    info.Initialize(new BinaryReader(new UnmanagedMemoryStream((byte*)(headerSize + physicalAddress), entryLength)));
+                    info.Initialize(new ReadOnlySpan<byte>((byte*)(headerSize + physicalAddress), entryLength));
                     if (info.CommitNum != long.MaxValue) continue;
 
                     // Otherwise, no more entries
@@ -379,7 +379,7 @@ public unsafe bool GetNext(MemoryPool<byte> pool, out IMemoryOwner<byte> entry,
                 if (isCommitRecord)
                 {
                     TsavoriteLogRecoveryInfo info = new();
-                    info.Initialize(new BinaryReader(new UnmanagedMemoryStream((byte*)physicalAddress, entryLength)));
+                    info.Initialize(new ReadOnlySpan<byte>((byte*)physicalAddress, entryLength));
                     if (info.CommitNum != long.MaxValue) continue;
 
                     // Otherwise, no more entries
@@ -442,7 +442,7 @@ public unsafe bool GetNext(MemoryPool<byte> pool, out IMemoryOwner<byte> entry,
                 if (isCommitRecord)
                 {
                     TsavoriteLogRecoveryInfo info = new();
-                    info.Initialize(new BinaryReader(new UnmanagedMemoryStream((byte*)physicalAddress, entryLength)));
+                    info.Initialize(new ReadOnlySpan<byte>((byte*)physicalAddress, entryLength));
                     if (info.CommitNum != long.MaxValue) continue;
 
                     // Otherwise, no more entries
@@ -590,7 +590,7 @@ public unsafe bool UnsafeGetNext(out byte* entry, out int entryLength, out long
                 if (isCommitRecord)
                 {
                     TsavoriteLogRecoveryInfo info = new();
-                    info.Initialize(new BinaryReader(new UnmanagedMemoryStream(entry, entryLength)));
+                    info.Initialize(new ReadOnlySpan<byte>(entry, entryLength));
                     if (info.CommitNum != long.MaxValue) continue;
 
                     // Otherwise, no more entries
@@ -715,12 +715,7 @@ internal unsafe bool ScanForwardForCommit(ref TsavoriteLogRecoveryInfo info, lon
                     if (!isCommitRecord) continue;
 
                     foundCommit = true;
-                    byte[] entry;
-                    // We allocate a byte array from heap
-                    entry = new byte[entryLength];
-                    fixed (byte* bp = entry)
-                        Buffer.MemoryCopy((void*)(headerSize + physicalAddress), bp, entryLength, entryLength);
-                    info.Initialize(new BinaryReader(new MemoryStream(entry)));
+                    info.Initialize(new ReadOnlySpan<byte>((void*)(headerSize + physicalAddress), entryLength));
 
                     Debug.Assert(info.CommitNum != -1);
 

diff --git a/libs/storage/Tsavorite/cs/src/core/TsavoriteLog/TsavoriteLogRecoveryInfo.cs b/libs/storage/Tsavorite/cs/src/core/TsavoriteLog/TsavoriteLogRecoveryInfo.cs
@@ -2,6 +2,7 @@
 // Licensed under the MIT license.
 
 using System;
+using System.Buffers.Binary;
 using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Diagnostics;
@@ -70,60 +71,65 @@ public void Initialize()
         /// <summary>
         /// Initialize from stream
         /// </summary>
-        /// <param name="reader"></param>
-        public void Initialize(BinaryReader reader)
+        /// <param name="input"></param>
+        public void Initialize(ReadOnlySpan<byte> input)
         {
-            int version;
-            long checkSum;
-            try
+            int version = BinaryPrimitives.ReadInt32LittleEndian(input);
+            input = input.Slice(sizeof(int));
+
+            long checkSum = BinaryPrimitives.ReadInt64LittleEndian(input);
+            input = input.Slice(sizeof(long));
+
+            BeginAddress = BinaryPrimitives.ReadInt64LittleEndian(input);
+            input = input.Slice(sizeof(long));
+
+            UntilAddress = BinaryPrimitives.ReadInt64LittleEndian(input);
+            input = input.Slice(sizeof(long));
+
+            if (version > 0)
             {
-                version = reader.ReadInt32();
-                checkSum = reader.ReadInt64();
-                BeginAddress = reader.ReadInt64();
-                UntilAddress = reader.ReadInt64();
-                if (version > 0)
-                    CommitNum = reader.ReadInt64();
-                else
-                    CommitNum = -1;
+                CommitNum = BinaryPrimitives.ReadInt64LittleEndian(input);
+                input = input.Slice(sizeof(long));
             }
-            catch (Exception e)
+            else
             {
-                throw new TsavoriteException("Unable to recover from previous commit. Inner exception: " + e.ToString());
+                CommitNum = -1;
             }
+
             if (version < 0 || version > TsavoriteLogRecoveryVersion)
                 throw new TsavoriteException("Invalid version found during commit recovery");
 
-            var iteratorCount = 0;
-            try
-            {
-                iteratorCount = reader.ReadInt32();
-            }
-            catch { }
+            if (BinaryPrimitives.TryReadInt32LittleEndian(input, out var iteratorCount))
+                input = input.Slice(sizeof(int));
 
             if (iteratorCount > 0)
             {
-                Iterators = new Dictionary<string, long>();
-                for (int i = 0; i < iteratorCount; i++)
+                Iterators = new Dictionary<string, long>(iteratorCount);
+                for (var i = 0; i < iteratorCount; i++)
                 {
-                    int len = reader.ReadInt32();
-                    byte[] bytes = reader.ReadBytes(len);
-                    Iterators.Add(Encoding.UTF8.GetString(bytes), reader.ReadInt64());
+                    var keyLength = BinaryPrimitives.ReadInt32LittleEndian(input);
+                    input = input.Slice(sizeof(int));
+
+                    var iteratorKey = Encoding.UTF8.GetString(input.Slice(0, keyLength));
+                    input = input.Slice(keyLength);
+
+                    var iteratorValue = BinaryPrimitives.ReadInt64LittleEndian(input);
+                    input = input.Slice(sizeof(long));
+
+                    Iterators.Add(iteratorKey, iteratorValue);
                 }
             }
 
             int cookieLength = -1;
             long cookieChecksum = 0;
             if (version >= TsavoriteLogRecoveryVersion)
             {
-                try
-                {
-                    cookieLength = reader.ReadInt32();
-                }
-                catch { }
+                if (BinaryPrimitives.TryReadInt32LittleEndian(input, out cookieLength))
+                    input = input.Slice(sizeof(int));
 
                 if (cookieLength >= 0)
                 {
-                    Cookie = reader.ReadBytes(cookieLength);
+                    Cookie = input.Slice(0, cookieLength).ToArray();
                     unsafe
                     {
                         fixed (byte* ptr = Cookie)