Merge branch 'master' into async-support

cs/src/core/Allocator/AllocatorBase.cs

@@ -1155,7 +1155,8 @@ private bool MonotonicUpdate(ref long variable, long newValue, out long oldValue
         /// </summary>
         /// <param name="tailAddress"></param>
         /// <param name="headAddress"></param>
-        public void RecoveryReset(long tailAddress, long headAddress)
+        /// <param name="beginAddress"></param>
+        public void RecoveryReset(long tailAddress, long headAddress, long beginAddress)
         {
             long tailPage = GetPage(tailAddress);
             long offsetInPage = GetOffsetInPage(tailAddress);
@@ -1168,7 +1169,8 @@ public void RecoveryReset(long tailAddress, long headAddress)
             var nextPageIndex = (pageIndex + 1) % BufferSize;
             if (tailAddress > 0)
                 AllocatePage(nextPageIndex);
-
+
+            BeginAddress = beginAddress;
             HeadAddress = headAddress;
             SafeHeadAddress = headAddress;
             FlushedUntilAddress = tailAddress;

cs/src/core/Index/Common/Contexts.cs

@@ -145,6 +145,10 @@ public struct HybridLogRecoveryInfo
         /// </summary>
         public long headAddress;
         /// <summary>
+        /// Begin address
+        /// </summary>
+        public long beginAddress;
+        /// <summary>
         /// Guid array
         /// </summary>
         public Guid[] guids;
@@ -215,6 +219,9 @@ public void Initialize(StreamReader reader)
             value = reader.ReadLine();
             headAddress = long.Parse(value);
 
+            value = reader.ReadLine();
+            beginAddress = long.Parse(value);
+
             value = reader.ReadLine();
             numThreads = int.Parse(value);
 
@@ -281,8 +288,9 @@ public byte[] ToByteArray()
                     writer.WriteLine(startLogicalAddress);
                     writer.WriteLine(finalLogicalAddress);
                     writer.WriteLine(headAddress);
-                    writer.WriteLine(checkpointTokens.Count);
+                    writer.WriteLine(beginAddress);
 
+                    writer.WriteLine(checkpointTokens.Count);
                     foreach (var kvp in checkpointTokens)
                     {
                         writer.WriteLine(kvp.Key);
@@ -315,6 +323,7 @@ public void DebugPrint()
             Debug.WriteLine("Start Logical Address: {0}", startLogicalAddress);
             Debug.WriteLine("Final Logical Address: {0}", finalLogicalAddress);
             Debug.WriteLine("Head Address: {0}", headAddress);
+            Debug.WriteLine("Begin Address: {0}", beginAddress);
             Debug.WriteLine("Num sessions recovered: {0}", numThreads);
             Debug.WriteLine("Recovered sessions: ");
             foreach (var sessionInfo in continueTokens)

cs/src/core/Index/Recovery/Checkpoint.cs

@@ -250,6 +250,7 @@ private bool GlobalMoveToNextState(SystemState currentState, SystemState nextSta
                             }
 
                             _hybridLogCheckpoint.info.headAddress = hlog.HeadAddress;
+                            _hybridLogCheckpoint.info.beginAddress = hlog.BeginAddress;
 
                             if (FoldOverSnapshot)
                             {

cs/src/core/Index/Recovery/Recovery.cs

@@ -121,16 +121,22 @@ private void InternalRecover(Guid indexToken, Guid hybridLogToken)
 
 
             // Read appropriate hybrid log pages into memory
-            RestoreHybridLog(recoveredHLCInfo.info.finalLogicalAddress, recoveredHLCInfo.info.headAddress);
+            RestoreHybridLog(recoveredHLCInfo.info.finalLogicalAddress, recoveredHLCInfo.info.headAddress, recoveredHLCInfo.info.beginAddress);
 
             // Recover session information
             _recoveredSessions = recoveredHLCInfo.info.continueTokens;
         }
 
-        private void RestoreHybridLog(long untilAddress, long headAddress)
+        private void RestoreHybridLog(long untilAddress, long headAddress, long beginAddress)
         {
-            // Special case: we do not load any records into memory
-            if (headAddress == untilAddress)
+            Debug.Assert(beginAddress <= headAddress);
+            Debug.Assert(headAddress <= untilAddress);
+
+            // Special cases: we do not load any records into memory
+            if (
+                (beginAddress == untilAddress) || // Empty log
+                ((headAddress == untilAddress) && (hlog.GetOffsetInPage(headAddress) == 0)) // Empty in-memory page
+                )
             {
                 hlog.AllocatePage(hlog.GetPageIndexForAddress(headAddress));
             }
@@ -171,7 +177,7 @@ private void RestoreHybridLog(long untilAddress, long headAddress)
                 }
             }
 
-            hlog.RecoveryReset(untilAddress, headAddress);
+            hlog.RecoveryReset(untilAddress, headAddress, beginAddress);
         }