Add ReadyToCompletePendingAsync for client to check if there is somet…

cs/src/core/ClientSession/ClientSession.cs

@@ -310,7 +310,8 @@ public void Refresh()
         }
 
         /// <summary>
-        /// Sync complete outstanding pending operations
+        /// Sync complete all outstanding pending operations
+        /// Async operations (ReadAsync) must be completed individually
         /// </summary>
         /// <param name="spinWait">Spin-wait for all pending operations on session to complete</param>
         /// <param name="spinWaitForCommit">Extend spin-wait until ongoing commit/checkpoint, if any, completes</param>
@@ -365,6 +366,22 @@ public async ValueTask CompletePendingAsync(bool waitForCommit = false, Cancella
                 await WaitForCommitAsync(token);
         }
 
+        /// <summary>
+        /// Check if at least one request is ready for CompletePending to be called on
+        /// Returns completed immediately if there are no outstanding requests
+        /// </summary>
+        /// <param name="token"></param>
+        /// <returns></returns>
+        public async ValueTask ReadyToCompletePendingAsync(CancellationToken token = default)
+        {
+            token.ThrowIfCancellationRequested();
+
+            if (fht.epoch.ThisInstanceProtected())
+                throw new NotSupportedException("Async operations not supported over protected epoch");
+
+            await fht.ReadyToCompletePendingAsync(this, token);
+        }
+
         /// <summary>
         /// Wait for commit of all operations completed until the current point in session.
         /// Does not itself issue checkpoint/commits.

cs/src/core/ClientSession/FASTERAsync.cs

@@ -26,6 +26,30 @@ public partial class FasterKV<Key, Value, Input, Output, Context, Functions> : F
         where Value : new()
         where Functions : IFunctions<Key, Value, Input, Output, Context>
     {
+
+        /// <summary>
+        /// Check if at least one request is ready for CompletePending to operate on
+        /// </summary>
+        /// <param name="clientSession"></param>
+        /// <param name="token"></param>
+        /// <returns></returns>
+        internal async ValueTask ReadyToCompletePendingAsync(ClientSession<Key, Value, Input, Output, Context, Functions> clientSession, CancellationToken token = default)
+        {
+            #region Previous pending requests
+            if (!RelaxedCPR)
+            {
+                if (clientSession.ctx.phase == Phase.IN_PROGRESS || clientSession.ctx.phase == Phase.WAIT_PENDING)
+                {
+                    if (clientSession.ctx.prevCtx.ioPendingRequests.Count != 0)
+                        await clientSession.ctx.prevCtx.readyResponses.WaitForEntryAsync();
+                }
+            }
+            #endregion
+
+            if (clientSession.ctx.ioPendingRequests.Count != 0)
+                await clientSession.ctx.readyResponses.WaitForEntryAsync();
+        }
+
         /// <summary>
         /// Complete outstanding pending operations that were issued synchronously
         /// Async operations (e.g., ReadAsync) need to be completed individually

cs/src/core/Utilities/AsyncQueue.cs

@@ -58,6 +58,16 @@ public async Task<T> DequeueAsync(CancellationToken cancellationToken = default)
             }
         }
 
+        /// <summary>
+        /// Wait for queue to have at least one entry
+        /// </summary>
+        /// <param name="token"></param>
+        /// <returns></returns>
+        public async Task WaitForEntryAsync(CancellationToken token = default)
+        {
+            await semaphore.WaitAsync(token);
+        }
+
         /// <summary>
         /// Try dequeue (if item exists)
         /// </summary>