Tiered storage device

cs/src/core/Allocator/AllocatorBase.cs

@@ -486,8 +486,8 @@ public AllocatorBase(LogSettings settings, IFasterEqualityComparer<Key> comparer
             else
                 this.epoch = epoch;
 
-            settings.LogDevice.Initialize(1L << settings.SegmentSizeBits);
-            settings.ObjectLogDevice?.Initialize(1L << settings.SegmentSizeBits);
+            settings.LogDevice.Initialize(1L << settings.SegmentSizeBits, epoch);
+            settings.ObjectLogDevice?.Initialize(1L << settings.SegmentSizeBits, epoch);
 
             // Page size
             LogPageSizeBits = settings.PageSizeBits;
@@ -870,7 +870,7 @@ public void ShiftReadOnlyToTail(out long tailAddress)
             tailAddress = GetTailAddress();
             long localTailAddress = tailAddress;
             long currentReadOnlyOffset = ReadOnlyAddress;
-            if (MonotonicUpdate(ref ReadOnlyAddress, tailAddress, out long oldReadOnlyOffset))
+            if (Utility.MonotonicUpdate(ref ReadOnlyAddress, tailAddress, out long oldReadOnlyOffset))
             {
                 epoch.BumpCurrentEpoch(() => OnPagesMarkedReadOnly(localTailAddress, false));
             }
@@ -882,7 +882,7 @@ public void ShiftReadOnlyToTail(out long tailAddress)
         /// <param name="newReadOnlyAddress"></param>
         public bool ShiftReadOnlyAddress(long newReadOnlyAddress)
         {
-            if (MonotonicUpdate(ref ReadOnlyAddress, newReadOnlyAddress, out long oldReadOnlyOffset))
+            if (Utility.MonotonicUpdate(ref ReadOnlyAddress, newReadOnlyAddress, out long oldReadOnlyOffset))
             {
                 epoch.BumpCurrentEpoch(() => OnPagesMarkedReadOnly(newReadOnlyAddress, false));
                 return true;
@@ -897,34 +897,33 @@ public bool ShiftReadOnlyAddress(long newReadOnlyAddress)
         public void ShiftBeginAddress(long newBeginAddress)
         {
             // First update the begin address
-            MonotonicUpdate(ref BeginAddress, newBeginAddress, out long oldBeginAddress);
+            Utility.MonotonicUpdate(ref BeginAddress, newBeginAddress, out long oldBeginAddress);
             // Then the head address
-            var h = MonotonicUpdate(ref HeadAddress, newBeginAddress, out long old);
+            var h = Utility.MonotonicUpdate(ref HeadAddress, newBeginAddress, out long old);
             // Finally the read-only address
-            var r = MonotonicUpdate(ref ReadOnlyAddress, newBeginAddress, out old);
+            var r = Utility.MonotonicUpdate(ref ReadOnlyAddress, newBeginAddress, out old);
 
             // Clean up until begin address
             epoch.BumpCurrentEpoch(() =>
             {
                 if (r)
                 {
-                    MonotonicUpdate(ref SafeReadOnlyAddress, newBeginAddress, out long _old);
-                    MonotonicUpdate(ref FlushedUntilAddress, newBeginAddress, out _old);
+                    Utility.MonotonicUpdate(ref SafeReadOnlyAddress, newBeginAddress, out long _old);
+                    Utility.MonotonicUpdate(ref FlushedUntilAddress, newBeginAddress, out _old);
                 }
                 if (h) OnPagesClosed(newBeginAddress);
 
-                DeleteAddressRange(oldBeginAddress, newBeginAddress);
+                TruncateUntilAddress(newBeginAddress);
             });
         }
 
         /// <summary>
-        /// Delete address range
+        /// Wraps <see cref="IDevice.TruncateUntilAddress(long)"/> when an allocator potentially has to interact with multiple devices
         /// </summary>
-        /// <param name="fromAddress"></param>
         /// <param name="toAddress"></param>
-        protected virtual void DeleteAddressRange(long fromAddress, long toAddress)
+        protected virtual void TruncateUntilAddress(long toAddress)
         {
-            device.DeleteAddressRange(fromAddress, toAddress);
+            device.TruncateUntilAddress(toAddress);
         }
 
         /// <summary>
@@ -935,7 +934,7 @@ protected virtual void DeleteAddressRange(long fromAddress, long toAddress)
         /// <param name="waitForPendingFlushComplete"></param>
         public void OnPagesMarkedReadOnly(long newSafeReadOnlyAddress, bool waitForPendingFlushComplete = false)
         {
-            if (MonotonicUpdate(ref SafeReadOnlyAddress, newSafeReadOnlyAddress, out long oldSafeReadOnlyAddress))
+            if (Utility.MonotonicUpdate(ref SafeReadOnlyAddress, newSafeReadOnlyAddress, out long oldSafeReadOnlyAddress))
             {
                 Debug.WriteLine("SafeReadOnly shifted from {0:X} to {1:X}", oldSafeReadOnlyAddress, newSafeReadOnlyAddress);
                 long startPage = oldSafeReadOnlyAddress >> LogPageSizeBits;
@@ -964,7 +963,7 @@ public void OnPagesMarkedReadOnly(long newSafeReadOnlyAddress, bool waitForPendi
         /// <param name="newSafeHeadAddress"></param>
         public void OnPagesClosed(long newSafeHeadAddress)
         {
-            if (MonotonicUpdate(ref SafeHeadAddress, newSafeHeadAddress, out long oldSafeHeadAddress))
+            if (Utility.MonotonicUpdate(ref SafeHeadAddress, newSafeHeadAddress, out long oldSafeHeadAddress))
             {
                 Debug.WriteLine("SafeHeadOffset shifted from {0:X} to {1:X}", oldSafeHeadAddress, newSafeHeadAddress);
 
@@ -1020,7 +1019,7 @@ private void PageAlignedShiftReadOnlyAddress(long currentTailAddress)
             long currentReadOnlyAddress = ReadOnlyAddress;
             long pageAlignedTailAddress = currentTailAddress & ~PageSizeMask;
             long desiredReadOnlyAddress = (pageAlignedTailAddress - ReadOnlyLagAddress);
-            if (MonotonicUpdate(ref ReadOnlyAddress, desiredReadOnlyAddress, out long oldReadOnlyAddress))
+            if (Utility.MonotonicUpdate(ref ReadOnlyAddress, desiredReadOnlyAddress, out long oldReadOnlyAddress))
             {
                 Debug.WriteLine("Allocate: Moving read-only offset from {0:X} to {1:X}", oldReadOnlyAddress, desiredReadOnlyAddress);
                 epoch.BumpCurrentEpoch(() => OnPagesMarkedReadOnly(desiredReadOnlyAddress));
@@ -1050,7 +1049,7 @@ private void PageAlignedShiftHeadAddress(long currentTailAddress)
             if (ReadCache && (newHeadAddress > HeadAddress))
                 EvictCallback(HeadAddress, newHeadAddress);
 
-            if (MonotonicUpdate(ref HeadAddress, newHeadAddress, out long oldHeadAddress))
+            if (Utility.MonotonicUpdate(ref HeadAddress, newHeadAddress, out long oldHeadAddress))
             {
                 Debug.WriteLine("Allocate: Moving head offset from {0:X} to {1:X}", oldHeadAddress, newHeadAddress);
                 epoch.BumpCurrentEpoch(() => OnPagesClosed(newHeadAddress));
@@ -1075,7 +1074,7 @@ public long ShiftHeadAddress(long desiredHeadAddress)
             if (ReadCache && (newHeadAddress > HeadAddress))
                 EvictCallback(HeadAddress, newHeadAddress);
 
-            if (MonotonicUpdate(ref HeadAddress, newHeadAddress, out long oldHeadAddress))
+            if (Utility.MonotonicUpdate(ref HeadAddress, newHeadAddress, out long oldHeadAddress))
             {
                 Debug.WriteLine("Allocate: Moving head offset from {0:X} to {1:X}", oldHeadAddress, newHeadAddress);
                 epoch.BumpCurrentEpoch(() => OnPagesClosed(newHeadAddress));
@@ -1104,33 +1103,8 @@ protected void ShiftFlushedUntilAddress()
 
             if (update)
             {
-                MonotonicUpdate(ref FlushedUntilAddress, currentFlushedUntilAddress, out long oldFlushedUntilAddress);
-            }
-        }
-
-
-
-        /// <summary>
-        /// Used by several functions to update the variable to newValue. Ignores if newValue is smaller or 
-        /// than the current value.
-        /// </summary>
-        /// <param name="variable"></param>
-        /// <param name="newValue"></param>
-        /// <param name="oldValue"></param>
-        /// <returns></returns>
-        private bool MonotonicUpdate(ref long variable, long newValue, out long oldValue)
-        {
-            oldValue = variable;
-            while (oldValue < newValue)
-            {
-                var foundValue = Interlocked.CompareExchange(ref variable, newValue, oldValue);
-                if (foundValue == oldValue)
-                {
-                    return true;
-                }
-                oldValue = foundValue;
+                Utility.MonotonicUpdate(ref FlushedUntilAddress, currentFlushedUntilAddress, out long oldFlushedUntilAddress);
             }
-            return false;
         }
 
         /// <summary>

cs/src/core/Allocator/BlittableAllocator.cs

@@ -153,11 +153,6 @@ protected override bool IsAllocated(int pageIndex)
             return values[pageIndex] != null;
         }
 
-        protected override void DeleteAddressRange(long fromAddress, long toAddress)
-        {
-            base.DeleteAddressRange(fromAddress, toAddress);
-        }
-
         protected override void WriteAsync<TContext>(long flushPage, IOCompletionCallback callback, PageAsyncFlushResult<TContext> asyncResult)
         {
             WriteAsync((IntPtr)pointers[flushPage % BufferSize],

cs/src/core/Allocator/GenericAllocator.cs

@@ -226,10 +226,10 @@ protected override bool IsAllocated(int pageIndex)
             return values[pageIndex] != null;
         }
 
-        protected override void DeleteAddressRange(long fromAddress, long toAddress)
+        protected override void TruncateUntilAddress(long toAddress)
         {
-            base.DeleteAddressRange(fromAddress, toAddress);
-            objectLogDevice.DeleteSegmentRange((int)(fromAddress >> LogSegmentSizeBits), (int)(toAddress >> LogSegmentSizeBits));
+            base.TruncateUntilAddress(toAddress);
+            objectLogDevice.TruncateUntilAddress(toAddress);
         }
 
         protected override void WriteAsync<TContext>(long flushPage, IOCompletionCallback callback,  PageAsyncFlushResult<TContext> asyncResult)

cs/src/core/Allocator/VarLenBlittableAllocator.cs

@@ -225,11 +225,6 @@ protected override bool IsAllocated(int pageIndex)
             return values[pageIndex] != null;
         }
 
-        protected override void DeleteAddressRange(long fromAddress, long toAddress)
-        {
-            base.DeleteAddressRange(fromAddress, toAddress);
-        }
-
         protected override void WriteAsync<TContext>(long flushPage, IOCompletionCallback callback, PageAsyncFlushResult<TContext> asyncResult)
         {
             WriteAsync((IntPtr)pointers[flushPage % BufferSize],

cs/src/core/Device/Devices.cs

@@ -13,14 +13,22 @@ namespace FASTER.core
     /// </summary>
     public static class Devices
     {
+        /// <summary>
+        /// This value is supplied for capacity when the device does not have a specified limit.
+        /// </summary>
+        public const long CAPACITY_UNSPECIFIED = -1;
+        private const string EMULATED_STORAGE_STRING = "UseDevelopmentStorage=true;";
+        private const string TEST_CONTAINER = "test";
+
         /// <summary>
         /// Create a storage device for the log
         /// </summary>
         /// <param name="logPath">Path to file that will store the log (empty for null device)</param>
         /// <param name="preallocateFile">Whether we try to preallocate the file on creation</param>
         /// <param name="deleteOnClose">Delete files on close</param>
+        /// <param name="capacity"></param>
         /// <returns>Device instance</returns>
-        public static IDevice CreateLogDevice(string logPath, bool preallocateFile = true, bool deleteOnClose = false)
+        public static IDevice CreateLogDevice(string logPath, bool preallocateFile = true, bool deleteOnClose = false, long capacity = CAPACITY_UNSPECIFIED)
         {
             if (string.IsNullOrWhiteSpace(logPath))
                 return new NullDevice();
@@ -30,12 +38,12 @@ public static IDevice CreateLogDevice(string logPath, bool preallocateFile = tru
 #if DOTNETCORE
             if (!RuntimeInformation.IsOSPlatform(OSPlatform.Windows))
             {
-                logDevice = new ManagedLocalStorageDevice(logPath, preallocateFile, deleteOnClose);
+                logDevice = new ManagedLocalStorageDevice(logPath, preallocateFile, deleteOnClose, capacity);
             }
             else
 #endif
             {
-                logDevice = new LocalStorageDevice(logPath, preallocateFile, deleteOnClose);
+                logDevice = new LocalStorageDevice(logPath, preallocateFile, deleteOnClose, capacity: capacity);
             }
             return logDevice;
         }

cs/src/core/Device/IDevice.cs

@@ -22,14 +22,44 @@ public interface IDevice
         string FileName { get; }
 
         /// <summary>
-        /// Initialize device
+        /// Returns the maximum capacity of the storage device, in number of bytes. 
+        /// If returned CAPACITY_UNSPECIFIED, the storage device has no specfied capacity limit. 
         /// </summary>
-        /// <param name="segmentSize"></param>
-        void Initialize(long segmentSize);
+        long Capacity { get; }
 
+        /// <summary>
+        /// A device breaks up each logical log into multiple self-contained segments that are of the same size.
+        /// It is an atomic unit of data that cannot be partially present on a device (i.e. either the entire segment
+        /// is present or no data from the segment is present). Examples of this include files or named blobs. This
+        /// property returns the size of each segment.
+        /// </summary>
+        long SegmentSize { get; }
 
-        /* Segmented addressing API */
+        /// <summary>
+        /// The index of the first segment present on this device
+        /// </summary>
+        int StartSegment { get; }
+
+        /// <summary>
+        /// The index of the last segment present on this device
+        /// </summary>
+        int EndSegment { get; }
+
+        /// <summary>
+        /// Initialize device. This function is used to pass optional information that may only be known after
+        /// FASTER initialization (whose constructor takes in IDevice upfront). Implementation are free to ignore
+        /// information if it does not need the supplied information.
+        /// 
+        /// This is a bit of a hack. 
+        /// </summary>
+        /// <param name="segmentSize"></param>
+        /// <param name="epoch">
+        /// The instance of the epoch protection framework to use, if needed
+        /// </param>
+        void Initialize(long segmentSize, LightEpoch epoch = null);
 
+
+        /* Segmented addressing API */
         /// <summary>
         /// Write
         /// </summary>
@@ -52,13 +82,6 @@ public interface IDevice
         /// <param name="asyncResult"></param>
         void ReadAsync(int segmentId, ulong sourceAddress, IntPtr destinationAddress, uint readLength, IOCompletionCallback callback, IAsyncResult asyncResult);
 
-        /// <summary>
-        /// Delete segment range
-        /// </summary>
-        /// <param name="fromSegment"></param>
-        /// <param name="toSegment"></param>
-        void DeleteSegmentRange(int fromSegment, int toSegment);
-
         /* Direct addressing API */
 
         /// <summary>
@@ -82,11 +105,51 @@ public interface IDevice
         void ReadAsync(ulong alignedSourceAddress, IntPtr alignedDestinationAddress, uint aligned_read_length, IOCompletionCallback callback, IAsyncResult asyncResult);
 
         /// <summary>
-        /// Delete address range
+        /// Truncates the log until the given address. The truncated portion should no longer be accessed as the device is no longer responsible for 
+        /// its maintenance, but physical deletion may not happen immediately.
+        /// </summary>
+        /// <param name="toAddress">upper bound of truncated address</param>
+        /// <param name="callback">callback to invoke when truncation is complete</param>
+        /// <param name="result">result to be passed to the callback</param>
+        void TruncateUntilAddressAsync(long toAddress, AsyncCallback callback, IAsyncResult result);
+
+        /// <summary>
+        /// Truncates the log until the given address. The truncated portion should no longer be accessed as the device is no longer responsible for 
+        /// its maintenance, but physical deletion may not happen immediately. This version of the function can block.
+        /// </summary>
+        /// <param name="toAddress">upper bound of truncated address</param>
+        void TruncateUntilAddress(long toAddress);
+
+        /// <summary>
+        /// Truncates the log until the given segment. Physical deletion of the given segments are guaranteed to have happened when the callback is invoked.
+        /// </summary>
+        /// <param name="toSegment">the largest (in index) segment to truncate</param>
+        /// <param name="callback">callback to invoke when truncation is complete</param>
+        /// <param name="result">result to be passed to the callback</param>
+        void TruncateUntilSegmentAsync(int toSegment, AsyncCallback callback, IAsyncResult result);
+
+        /// <summary>
+        /// Truncates the log until the given segment. Physical deletion of the given segments are guaranteed to have happened when the function returns.
+        /// This version of the function can block.
+        /// </summary>
+        /// <param name="toSegment">the largest (in index) segment to truncate</param>
+        void TruncateUntilSegment(int toSegment);
+
+        /// <summary>
+        /// Removes a single segment from the device. This function should not normally be called.
+        /// Instead, use <see cref="TruncateUntilAddressAsync(long, AsyncCallback, IAsyncResult)"/>
+        /// </summary>
+        /// <param name="segment">index of the segment to remov</param>
+        /// <param name="callback">callback to invoke when removal is complete</param>
+        /// <param name="result">result to be passed to the callback</param>
+        void RemoveSegmentAsync(int segment, AsyncCallback callback, IAsyncResult result);
+
+        /// <summary>
+        /// Removes a single segment from the device. This function should not normally be called.
+        /// Instead, use <see cref="TruncateUntilAddressAsync(long, AsyncCallback, IAsyncResult)"/>
         /// </summary>
-        /// <param name="fromAddress"></param>
-        /// <param name="toAddress"></param>
-        void DeleteAddressRange(long fromAddress, long toAddress);
+        /// <param name="segment">index of the segment to remov</param>
+        void RemoveSegment(int segment);
 
         /* Close */