NCBC-4005 Implement extParallelUnstaging

Motivation
==========
Unstaging in parallel will significantly improve performance of
transactions which have many staged mutation.

Modification
============
Do the unstaging in parallel.  Limit the number of unstaging docs
that are in-flight at any time to an arbitrary number of 100 for
now, but that constant can be changed if needed.   We keep up to
100 unstaging tasks in parallel, which sounds sufficient.   Note
this is only when they are done entirely with KV - a query in the
transaction then makes the query server do the commit.

Results
=======
The extention's tests pass, no alarming failures elsewhere.

Change-Id: I1f7515cc99e008cb8adf1bf2844d0a5fbd5c9a95
Reviewed-on: https://review.couchbase.org/c/couchbase-net-client/+/226777
Reviewed-by: Michael Reiche <michael.reiche@couchbase.com>
Tested-by: Build Bot <build@couchbase.com>

Author: davidkelly

src/Couchbase/Client/Transactions/AttemptContext.cs

@@ -53,6 +53,7 @@ public class AttemptContext
         private readonly TransactionContext _overallContext;
         private readonly MergedTransactionConfig _config;
         private readonly ITestHooks _testHooks;
+        private readonly int _unstagingConcurrency = 100;
         internal IRedactor Redactor { get; }
         internal AttemptStates AttemptState = AttemptStates.NOTHING_WRITTEN;
         private readonly ErrorTriage _triage;
@@ -1272,26 +1273,43 @@ private async Task UnstageDocs(IRequestSpan? parentSpan)
         {
             using var traceSpan = TraceSpan(parent: parentSpan);
             var allStagedMutations = _stagedMutations.ToList();
-            foreach (var sm in allStagedMutations)
+            var sem = new SemaphoreSlim(_unstagingConcurrency);
+            var tasks = allStagedMutations.Select(async sm =>
             {
+                await UnstageDoc(sm, sem).CAF();
+            });
+            await Task.WhenAll(tasks).CAF();
+        }
+
+        private async Task UnstageDoc(StagedMutation sm, SemaphoreSlim sem)
+        {
+            try
+            {
+                await sem.WaitAsync().CAF();
                 (var cas, var content) = await FetchIfNeededBeforeUnstage(sm).CAF();
                 switch (sm.Type)
                 {
                     case StagedMutationType.Remove:
                         await UnstageRemove(sm).CAF();
                         break;
                     case StagedMutationType.Insert:
-                        await UnstageInsertOrReplace(sm, cas, content, insertMode: true, ambiguityResolutionMode: false)
+                        await UnstageInsertOrReplace(sm, cas, content, insertMode: true,
+                                ambiguityResolutionMode: false)
                             .CAF();
                         break;
                     case StagedMutationType.Replace:
                         await UnstageInsertOrReplace(sm, cas, content, insertMode: false,
                             ambiguityResolutionMode: false).CAF();
                         break;
                     default:
-                        throw new InvalidOperationException($"Cannot un-stage transaction mutation of type {sm.Type}");
+                        throw new InvalidOperationException(
+                            $"Cannot un-stage transaction mutation of type {sm.Type}");
                 }
             }
+            finally
+            {
+                sem.Release();
+            }
         }
 
         private async Task UnstageRemove(StagedMutation sm, bool ambiguityResolutionMode = false,
@@ -1848,24 +1866,33 @@ internal async Task RollbackWithKv(bool isAppRollback, IRequestSpan? parentSpan)
 
             await SetAtrAborted(isAppRollback, traceSpan.Item).CAF();
             var allMutations = _stagedMutations.ToList();
-            foreach (var sm in allMutations)
+            var sem = new SemaphoreSlim(_unstagingConcurrency);
+            var tasks = allMutations.Select(async sm =>
             {
-                switch (sm.Type)
+                try
                 {
-                    case StagedMutationType.Insert:
-                        await RollbackStagedInsert(sm, traceSpan.Item).CAF();
-                        break;
-                    case StagedMutationType.Remove:
-                    case StagedMutationType.Replace:
-                        await RollbackStagedReplaceOrRemove(sm, traceSpan.Item).CAF();
-                        break;
-                    default:
-                        throw new InvalidOperationException(sm.Type +
-                                                            " is not a supported mutation type for rollback.");
+                    await sem.WaitAsync().CAF();
+                    switch (sm.Type)
+                    {
+                        case StagedMutationType.Insert:
+                            await RollbackStagedInsert(sm, traceSpan.Item).CAF();
+                            break;
+                        case StagedMutationType.Remove:
+                        case StagedMutationType.Replace:
+                            await RollbackStagedReplaceOrRemove(sm, traceSpan.Item).CAF();
+                            break;
+                        default:
+                            throw new InvalidOperationException(sm.Type +
+                                                                " is not a supported mutation type for rollback.");
 
+                    }
                 }
-            }
-
+                finally
+                {
+                    sem.Release();
+                }
+            });
+            await  Task.WhenAll(tasks).CAF();
             await SetAtrRolledBack(traceSpan.Item).CAF();
         }
 

src/Couchbase/Client/Transactions/ProtocolVersion.cs

@@ -33,6 +33,7 @@ public static IEnumerable<ExtensionName> ExtensionsSupported()
             yield return new ExtensionName("ExtThreadSafe", "EXT_THREAD_SAFE", "TS");
             yield return new ExtensionName("ExtInsertExisting", "EXT_INSERT_EXISTING", "IX");
             yield return new ExtensionName("ExtReplaceBodyWithXattr", "EXT_REPLACE_BODY_WITH_XATTR","RX");
+            yield return new ExtensionName("ExtParallelUnstaging", "EXT_PARALLEL_UNSTAGING", "PU");
         }
 
         internal static bool Supported(string shortCode) => SupportedShortCodes.Value.Contains(shortCode);