Decrement persistence thread merge counter when syncronous processing is complete

Add a generic interface for letting an operation know that the synchronous
parts of its processing in the persistence thread is complete. This allows
a potentially longer-running async operation to free up any limits that
were put in place when it was taking up synchronous thread resources.

Currently only used by merge-related operations (that may dispatch many
async ops). Since we have a max upper bound for how many threads in a stripe
may be processing merge ops at the same time (to avoid blocking client ops),
we previously could effectively stall the pipelining of merges caused by
hitting the concurrency limit even if all persistence threads were otherwise
idle (waiting for prior async merge ops to complete).
We now explicitly decrease the merge concurrency counter once the synchronous
processing is done, allowing us to take on further merges immediately.

1 files changed, 4 insertions, 0 deletions

diff --git a/storage/src/tests/persistence/persistencetestutils.h b/storage/src/tests/persistence/persistencetestutils.h
index fc986c3c6f2..4bbff9bb2ca 100644
--- a/storage/src/tests/persistence/persistencetestutils.h
+++ b/storage/src/tests/persistence/persistencetestutils.h
@@ -50,6 +50,8 @@ public:
         api::LockingRequirements lockingRequirements() const noexcept override {
             return api::LockingRequirements::Shared;
         }
+        void signal_operation_sync_phase_done() noexcept override {}
+        bool wants_sync_phase_done_notification() const noexcept override { return false; }
         static std::shared_ptr<NoBucketLock> make(document::Bucket bucket) {
             return std::make_shared<NoBucketLock>(bucket);
         }
@@ -78,6 +80,8 @@ public:
         api::LockingRequirements lockingRequirements() const noexcept override {
             return api::LockingRequirements::Exclusive;
         }
+        void signal_operation_sync_phase_done() noexcept override {}
+        bool wants_sync_phase_done_notification() const noexcept override { return false; }
         static std::shared_ptr<MockBucketLock> make(document::Bucket bucket, MockBucketLocks& locks) {
             return std::make_shared<MockBucketLock>(bucket, locks);
         }