[C#] FasterLog Commit Optimization (v2)

cs/samples/FasterLogPubSub/Program.cs

@@ -78,7 +78,7 @@ static async Task CommitterAsync(FasterLog log, CancellationToken cancellationTo
 
                     Console.WriteLine("Committing...");
 
-                    await log.CommitAsync(cancellationToken);
+                    await log.CommitAsync(token: cancellationToken);
                 }
             }
             catch (OperationCanceledException) { }

cs/src/core/Allocator/AllocatorBase.cs

@@ -1762,8 +1762,12 @@ public void AsyncFlushPages(long fromAddress, long untilAddress)
                     }
                     else
                     {
+                        // Because we are invoking the callback away from the usual codepath, need to externally
+                        // ensure that flush address are updated in order
+                        while (FlushedUntilAddress < asyncResult.fromAddress) Thread.Yield();
                         // Could not add to pending flush list, treat as a failed write
-                        AsyncFlushPageCallback(1, 0, asyncResult);
+                        // Use a special errorCode to convey that this is not from a syscall
+                        AsyncFlushPageCallback(16000, 0, asyncResult);
                     }
                 }
                 else

cs/src/core/Allocator/WorkQueueLIFO.cs

@@ -3,6 +3,8 @@
 
 using System;
 using System.Collections.Concurrent;
+using System.Runtime.CompilerServices;
+using System.Runtime.InteropServices;
 using System.Threading;
 using System.Threading.Tasks;
 
@@ -12,18 +14,30 @@ namespace FASTER.core
     /// Shared work queue that ensures one worker at any given time. Uses LIFO ordering of work.
     /// </summary>
     /// <typeparam name="T"></typeparam>
-    class WorkQueueLIFO<T>
+    class WorkQueueLIFO<T> : IDisposable
     {
         const int kMaxQueueSize = 1 << 30;
         readonly ConcurrentStack<T> _queue;
         readonly Action<T> _work;
-        int _count;
+        private int _count;
+        private bool _disposed;
 
         public WorkQueueLIFO(Action<T> work)
         {
             _queue = new ConcurrentStack<T>();
             _work = work;
             _count = 0;
+            _disposed = false;
+        }
+
+        public void Dispose()
+        {
+            _disposed = true;
+            // All future enqueue requests will no longer perform work after _disposed is set to true.
+            while (_count != 0)
+                Thread.Yield();
+            // After this point, any previous work must have completed. Even if another enqueue request manipulates the
+            // count field, they are guaranteed to see disposed and not enqueue any actual work.
         }
 
         /// <summary>
@@ -32,16 +46,24 @@ public WorkQueueLIFO(Action<T> work)
         /// </summary>
         /// <param name="work">Work to enqueue</param>
         /// <param name="asTask">Process work as separate task</param>
-        public void EnqueueAndTryWork(T work, bool asTask)
+        /// <returns> whether the enqueue is successful. Enqueuing into a disposed WorkQueue will fail and the task will not be performed</returns>>
+        public bool EnqueueAndTryWork(T work, bool asTask)
         {
             Interlocked.Increment(ref _count);
+            if (_disposed)
+            {
+                // Remove self from count in case Dispose() is actively waiting for completion
+                Interlocked.Decrement(ref _count);
+                return false;
+            }
+
             _queue.Push(work);
 
             // Try to take over work queue processing if needed
             while (true)
             {
                 int count = _count;
-                if (count >= kMaxQueueSize) return;
+                if (count >= kMaxQueueSize) return true;
                 if (Interlocked.CompareExchange(ref _count, count + kMaxQueueSize, count) == count)
                     break;
             }
@@ -50,6 +72,7 @@ public void EnqueueAndTryWork(T work, bool asTask)
                 _ = Task.Run(() => ProcessQueue());
             else
                 ProcessQueue();
+            return true;
         }
 
         private void ProcessQueue()
@@ -59,12 +82,12 @@ private void ProcessQueue()
             {
                 while (_queue.TryPop(out var workItem))
                 {
-                    Interlocked.Decrement(ref _count);
                     try
                     {
                         _work(workItem);
                     }
                     catch { }
+                    Interlocked.Decrement(ref _count);
                 }
 
                 int count = _count;

cs/src/core/Epochs/LightEpoch.cs

@@ -345,8 +345,8 @@ public void BumpCurrentEpoch(Action onDrain)
                     i = 0;
                     if (++j == 500)
                     {
+                        // Spin until there is a free entry in the drain list
                         j = 0;
-                        Debug.WriteLine("Delay finding a free entry in the drain list");
                     }
                 }
             }

cs/src/core/Index/CheckpointManagement/DeviceLogCommitCheckpointManager.cs

@@ -29,11 +29,6 @@ public class DeviceLogCommitCheckpointManager : ILogCommitManager, ICheckpointMa
         private IDevice singleLogCommitDevice;
         private bool _disposed;
 
-        /// <summary>
-        /// Next commit number
-        /// </summary>
-        private long commitNum;
-
         /// <summary>
         /// Track historical commits for automatic purging
         /// </summary>
@@ -52,7 +47,6 @@ public DeviceLogCommitCheckpointManager(INamedDeviceFactory deviceFactory, IChec
             this.deviceFactory = deviceFactory;
             this.checkpointNamingScheme = checkpointNamingScheme;
 
-            this.commitNum = 0;
             this.semaphore = new SemaphoreSlim(0);
 
             this.overwriteLogCommits = overwriteLogCommits;
@@ -99,9 +93,12 @@ public DeviceLogCommitCheckpointManager(INamedDeviceFactory deviceFactory, strin
         #region ILogCommitManager
 
         /// <inheritdoc />
-        public unsafe void Commit(long beginAddress, long untilAddress, byte[] commitMetadata)
+        public bool PreciseCommitNumRecoverySupport() => !overwriteLogCommits;
+
+        /// <inheritdoc />
+        public unsafe void Commit(long beginAddress, long untilAddress, byte[] commitMetadata, long commitNum)
         {
-            var device = NextCommitDevice();
+            var device = NextCommitDevice(commitNum);
 
             if (device == null) return;
 
@@ -135,6 +132,21 @@ public IEnumerable<long> ListCommits()
             return deviceFactory.ListContents(checkpointNamingScheme.FasterLogCommitBasePath()).Select(e => checkpointNamingScheme.CommitNumber(e)).OrderByDescending(e => e);
         }
 
+        /// <inheritdoc />
+        public void RemoveCommit(long commitNum)
+        {
+            if (overwriteLogCommits)
+                throw new FasterException("removing commit by commit num is not supported when overwriting log commits");
+            deviceFactory.Delete(checkpointNamingScheme.FasterLogCommitMetadata(commitNum));
+        }
+
+        /// <inheritdoc />
+        public void RemoveAllCommits()
+        {
+            foreach (var commitNum in ListCommits())
+                RemoveCommit(commitNum);
+        }
+
         /// <inheritdoc />
         public byte[] GetCommitMetadata(long commitNum)
         {
@@ -156,7 +168,6 @@ public byte[] GetCommitMetadata(long commitNum)
             else
             {
                 device = deviceFactory.Get(checkpointNamingScheme.FasterLogCommitMetadata(commitNum));
-                this.commitNum = commitNum + 1;
             }
             if (device == null) return null;
 
@@ -176,14 +187,14 @@ public byte[] GetCommitMetadata(long commitNum)
             return new Span<byte>(body).Slice(sizeof(int)).ToArray();
         }
 
-        private IDevice NextCommitDevice()
+        private IDevice NextCommitDevice(long commitNum)
         {
             if (overwriteLogCommits)
             {
                 if (_disposed) return null;
                 if (singleLogCommitDevice == null)
                 {
-                    singleLogCommitDevice = deviceFactory.Get(checkpointNamingScheme.FasterLogCommitMetadata(commitNum));
+                    singleLogCommitDevice = deviceFactory.Get(checkpointNamingScheme.FasterLogCommitMetadata(0));
                     if (_disposed)
                     {
                         singleLogCommitDevice?.Dispose();
@@ -194,11 +205,11 @@ private IDevice NextCommitDevice()
                 return singleLogCommitDevice;
             }
 
-            return deviceFactory.Get(checkpointNamingScheme.FasterLogCommitMetadata(commitNum++));
+            var result =  deviceFactory.Get(checkpointNamingScheme.FasterLogCommitMetadata(commitNum));
+            return result;
         }
         #endregion
-
-
+
         #region ICheckpointManager
         /// <inheritdoc />
         public unsafe void CommitIndexCheckpoint(Guid indexToken, byte[] commitMetadata)