issue-5432: drain async handle operations queue back to back

cloud/filestore/config/filesystem.proto

@@ -40,8 +40,8 @@ message TFileSystemConfig
     // Async processing of destroy handle requests.
     optional bool AsyncDestroyHandleEnabled = 11;
 
-    // Period of processing create/destroy handle requests.
-    optional uint32 AsyncHandleOperationPeriod = 12;
+    // Polling period when the async handle operations queue is empty.
+    optional uint32 AsyncHandleOperationIdlePeriod = 12;
 
     // Enable O_DIRECT when working with files
     optional bool DirectIoEnabled = 13;
@@ -133,4 +133,7 @@ message TFileSystemConfig
 
     // Availability interval duration (in ms); 0 means the default.
     optional uint32 AvailabilityTrackingInterval = 39;
+
+    // Delay between processing entries in the async handle operations queue.
+    optional uint32 AsyncHandleOperationDrainPeriod = 40;
 }

cloud/filestore/config/server.proto

@@ -145,8 +145,8 @@ message TLocalServiceConfig
     // Async processing of destroy handle requests
     optional bool AsyncDestroyHandleEnabled = 12;
 
-    // Period of processing create/destroy handle requests
-    optional uint32 AsyncHandleOperationPeriod = 13;
+    // Polling period when the async handle operations queue is empty.
+    optional uint32 AsyncHandleOperationIdlePeriod = 13;
 
     // Use open_by_handle_at to lookup nodes
     optional bool OpenNodeByHandleEnabled = 14;
@@ -234,6 +234,9 @@ message TLocalServiceConfig
 
     // Async processing of destroy handle requests only for read only handles.
     optional bool AsyncDestroyReadOnlyHandleEnabled = 42;
+
+    // Delay between processing entries in the async handle operations queue.
+    optional uint32 AsyncHandleOperationDrainPeriod = 43;
 }
 
 ////////////////////////////////////////////////////////////////////////////////

cloud/filestore/config/storage.proto

@@ -400,8 +400,8 @@ message TStorageConfig
     // Async processing of destroy handle requests.
     optional bool AsyncDestroyHandleEnabled = 379;
 
-    // Period of processing create/destroy handle requests.
-    optional uint32 AsyncHandleOperationPeriod = 380;
+    // Polling period when the async handle operations queue is empty, in ms.
+    optional uint32 AsyncHandleOperationIdlePeriod = 380;
 
     // Dynamic node registration params.
     optional uint32 NodeRegistrationMaxAttempts = 381;
@@ -902,4 +902,8 @@ message TStorageConfig
 
     // Availability interval duration (in ms); 0 means the default.
     optional uint32 AvailabilityTrackingInterval = 540;
+
+    // Delay between processing entries in the async handle operations queue,
+    // in ms.
+    optional uint32 AsyncHandleOperationDrainPeriod = 541;
 }

cloud/filestore/libs/service_local/config.cpp

@@ -29,7 +29,8 @@ namespace {
     xxx(GuestWriteBackCacheEnabled,  bool,          false                     )\
     xxx(AsyncDestroyHandleEnabled,   bool,          false                     )\
     xxx(AsyncDestroyReadOnlyHandleEnabled, bool,    false                     )\
-    xxx(AsyncHandleOperationPeriod,  TDuration,     50ms                      )\
+    xxx(AsyncHandleOperationIdlePeriod,  TDuration, 50ms                      )\
+    xxx(AsyncHandleOperationDrainPeriod, TDuration, 0ms                       )\
     xxx(OpenNodeByHandleEnabled,     bool,          false                     )\
     xxx(NodeCleanupBatchSize,        ui32,          1000                      )\
     xxx(ZeroCopyEnabled,             bool,          false                     )\

cloud/filestore/libs/service_local/config.h

@@ -107,7 +107,8 @@ class TLocalFileStoreConfig
 
     bool GetAsyncDestroyHandleEnabled() const;
     bool GetAsyncDestroyReadOnlyHandleEnabled() const;
-    TDuration GetAsyncHandleOperationPeriod() const;
+    TDuration GetAsyncHandleOperationIdlePeriod() const;
+    TDuration GetAsyncHandleOperationDrainPeriod() const;
 
     bool GetOpenNodeByHandleEnabled() const;
 

cloud/filestore/libs/service_local/fs_session.cpp

@@ -43,8 +43,10 @@ NProto::TCreateSessionResponse TLocalFileSystem::CreateSession(
             Config->GetAsyncDestroyHandleEnabled());
         features->SetAsyncDestroyReadOnlyHandleEnabled(
             Config->GetAsyncDestroyReadOnlyHandleEnabled());
-        features->SetAsyncHandleOperationPeriod(
-            Config->GetAsyncHandleOperationPeriod().MilliSeconds());
+        features->SetAsyncHandleOperationIdlePeriod(
+            Config->GetAsyncHandleOperationIdlePeriod().MilliSeconds());
+        features->SetAsyncHandleOperationDrainPeriod(
+            Config->GetAsyncHandleOperationDrainPeriod().MilliSeconds());
         // The local service publishes only the legacy ZeroCopyEnabled feature,
         // which the FUSE config translates into both ZeroCopyReadEnabled and
         // ZeroCopyWriteEnabled (see BuildFileSystemConfig in vfs_fuse/loop.cpp).

cloud/filestore/libs/storage/core/config.cpp

@@ -264,7 +264,10 @@ using TAliases = NProto::TStorageConfig::TFilestoreAliases;
     xxx(AsyncCreateHandleEnabled,                       bool,       false     )\
     xxx(TabletUnsafeAsyncReadOnlyCreateHandleEnabled,   bool,       false     )\
     xxx(TabletUnsafeAsyncDestroyHandleEnabled,          bool,       false     )\
-    xxx(AsyncHandleOperationPeriod,    TDuration,  TDuration::MilliSeconds(50))\
+    xxx(AsyncHandleOperationIdlePeriod,                                        \
+                                       TDuration,  TDuration::MilliSeconds(50))\
+    xxx(AsyncHandleOperationDrainPeriod,                                       \
+                                       TDuration,  TDuration::Zero()          )\
     xxx(UnconfirmedCreateHandleGraceTimeout,                                   \
                                        TDuration,  TDuration::Minutes(2)      )\
                                                                                \

cloud/filestore/libs/storage/core/config.h

@@ -277,7 +277,8 @@ class TStorageConfig
     bool GetAsyncCreateHandleEnabled() const;
     bool GetTabletUnsafeAsyncReadOnlyCreateHandleEnabled() const;
     bool GetTabletUnsafeAsyncDestroyHandleEnabled() const;
-    TDuration GetAsyncHandleOperationPeriod() const;
+    TDuration GetAsyncHandleOperationIdlePeriod() const;
+    TDuration GetAsyncHandleOperationDrainPeriod() const;
     TDuration GetUnconfirmedCreateHandleGraceTimeout() const;
 
     void Dump(IOutputStream& out) const;

cloud/filestore/libs/storage/tablet/tablet_actor_createsession.cpp

@@ -52,8 +52,10 @@ void FillFeatures(
         config.GetAsyncDestroyReadOnlyHandleEnabled());
     features->SetAsyncCreateHandleEnabled(
         config.GetAsyncCreateHandleEnabled());
-    features->SetAsyncHandleOperationPeriod(
-        config.GetAsyncHandleOperationPeriod().MilliSeconds());
+    features->SetAsyncHandleOperationIdlePeriod(
+        config.GetAsyncHandleOperationIdlePeriod().MilliSeconds());
+    features->SetAsyncHandleOperationDrainPeriod(
+        config.GetAsyncHandleOperationDrainPeriod().MilliSeconds());
 
     features->SetGuestWriteBackCacheEnabled(
         config.GetGuestWriteBackCacheEnabled());

cloud/filestore/libs/storage/tablet/tablet_ut_sessions.cpp

@@ -1420,11 +1420,11 @@ Y_UNIT_TEST_SUITE(TIndexTabletTest_Sessions)
         NProto::TFileStoreFeatures features;
         features.SetThreeStageWriteThreshold(64_KB);
         features.SetPreferredBlockSize(4_KB);
-        features.SetAsyncHandleOperationPeriod(
-            TDuration::MilliSeconds(50).MilliSeconds());
         features.SetHasXAttrs(true);
         features.SetMaxFuseLoopThreads(1);
         features.SetTabletDirectRdmaEnabled(false);
+        features.SetAsyncHandleOperationIdlePeriod(
+            TDuration::MilliSeconds(50).MilliSeconds());
 
         DoTestShouldReturnFeaturesInCreateSessionResponse(config, features);
 
@@ -1438,8 +1438,10 @@ Y_UNIT_TEST_SUITE(TIndexTabletTest_Sessions)
         config.SetPreferredBlockSizeMultiplier(2);
         config.SetAsyncDestroyHandleEnabled(true);
         config.SetAsyncDestroyReadOnlyHandleEnabled(true);
-        config.SetAsyncHandleOperationPeriod(
+        config.SetAsyncHandleOperationIdlePeriod(
             TDuration::MilliSeconds(100).MilliSeconds());
+        config.SetAsyncHandleOperationDrainPeriod(
+            TDuration::MilliSeconds(10).MilliSeconds());
         config.SetGuestPageCacheDisabled(true);
         config.SetExtendedAttributesDisabled(true);
         config.SetServerWriteBackCacheEnabled(true);
@@ -1473,8 +1475,10 @@ Y_UNIT_TEST_SUITE(TIndexTabletTest_Sessions)
         features.SetPreferredBlockSize(4_KB * 2);
         features.SetAsyncDestroyHandleEnabled(true);
         features.SetAsyncDestroyReadOnlyHandleEnabled(true);
-        features.SetAsyncHandleOperationPeriod(
+        features.SetAsyncHandleOperationIdlePeriod(
             TDuration::MilliSeconds(100).MilliSeconds());
+        features.SetAsyncHandleOperationDrainPeriod(
+            TDuration::MilliSeconds(10).MilliSeconds());
         features.SetGuestPageCacheDisabled(true);
         features.SetExtendedAttributesDisabled(true);
         features.SetServerWriteBackCacheEnabled(true);

cloud/filestore/libs/vfs_fuse/config.cpp

@@ -32,7 +32,8 @@ namespace {
     xxx(AsyncDestroyHandleEnabled,          bool,       false                 )\
     xxx(AsyncDestroyReadOnlyHandleEnabled,  bool,       false                 )\
     xxx(AsyncCreateHandleEnabled,           bool,       false                 )\
-    xxx(AsyncHandleOperationPeriod, TDuration,  TDuration::MilliSeconds(50)   )\
+    xxx(AsyncHandleOperationIdlePeriod, TDuration, TDuration::MilliSeconds(50))\
+    xxx(AsyncHandleOperationDrainPeriod, TDuration, TDuration::Zero()         )\
                                                                                \
     xxx(DirectIoEnabled,            bool,       false                         )\
     xxx(DirectIoAlign,              ui32,       4_KB                          )\

cloud/filestore/libs/vfs_fuse/config.h

@@ -41,7 +41,8 @@ struct TFileSystemConfig
     bool GetAsyncDestroyHandleEnabled() const;
     bool GetAsyncDestroyReadOnlyHandleEnabled() const;
     bool GetAsyncCreateHandleEnabled() const;
-    TDuration GetAsyncHandleOperationPeriod() const;
+    TDuration GetAsyncHandleOperationIdlePeriod() const;
+    TDuration GetAsyncHandleOperationDrainPeriod() const;
 
     bool GetDirectIoEnabled() const;
     ui32 GetDirectIoAlign() const;

cloud/filestore/libs/vfs_fuse/fs_impl.cpp

@@ -88,7 +88,8 @@ TFileSystem::~TFileSystem()
 void TFileSystem::Init()
 {
     STORAGE_INFO("scheduling handle ops queue processing");
-    ScheduleProcessHandleOpsQueue();
+    ScheduleProcessHandleOpsQueue(
+        Config->GetAsyncHandleOperationDrainPeriod());
 }
 
 void TFileSystem::Reset()
@@ -97,11 +98,11 @@ void TFileSystem::Reset()
     DirectoryHandleCache->Reset();
 }
 
-void TFileSystem::ScheduleProcessHandleOpsQueue()
+void TFileSystem::ScheduleProcessHandleOpsQueue(TDuration delay)
 {
     if (HandleOpsQueue) {
         Scheduler->Schedule(
-            Timer->Now() + Config->GetAsyncHandleOperationPeriod(),
+            Timer->Now() + delay,
             [=, ptr = weak_from_this()]()
             {
                 if (auto self = ptr.lock()) {
@@ -415,7 +416,8 @@ void TFileSystem::CompleteHandleOpsQueueEntry()
         HandleOpsQueue->PopFront();
     }
     ProcessDelayedRelease();
-    ScheduleProcessHandleOpsQueue();
+    ScheduleProcessHandleOpsQueue(
+        Config->GetAsyncHandleOperationDrainPeriod());
 }
 
 void TFileSystem::ProcessDelayedRelease()
@@ -440,7 +442,8 @@ void TFileSystem::ProcessHandleOpsQueue()
 {
     TGuard g{HandleOpsQueueLock};
     if (HandleOpsQueue->Empty()) {
-        ScheduleProcessHandleOpsQueue();
+        ScheduleProcessHandleOpsQueue(
+            Config->GetAsyncHandleOperationIdlePeriod());
         return;
     }
 
@@ -451,7 +454,8 @@ void TFileSystem::ProcessHandleOpsQueue()
             << "Failed to get TQueueEntry from queue, filesystem: "
             << Config->GetFileSystemId());
         HandleOpsQueue->PopFront();
-        ScheduleProcessHandleOpsQueue();
+        ScheduleProcessHandleOpsQueue(
+            Config->GetAsyncHandleOperationIdlePeriod());
         return;
     }
 
@@ -513,7 +517,8 @@ void TFileSystem::ProcessHandleOpsQueue()
             TStringBuilder() << "Unexpected TQueueEntry in queue, filesystem: "
                              << Config->GetFileSystemId());
         HandleOpsQueue->PopFront();
-        ScheduleProcessHandleOpsQueue();
+        ScheduleProcessHandleOpsQueue(
+            Config->GetAsyncHandleOperationIdlePeriod());
         return;
     }
 

cloud/filestore/libs/vfs_fuse/fs_impl.h

@@ -514,7 +514,7 @@ class TFileSystem final
 
     void ClearDirectoryCache();
 
-    void ScheduleProcessHandleOpsQueue();
+    void ScheduleProcessHandleOpsQueue(TDuration delay);
     void ProcessHandleOpsQueue();
 
     void DoWrite(

cloud/filestore/libs/vfs_fuse/fs_ut.cpp

@@ -3235,6 +3235,112 @@ Y_UNIT_TEST_SUITE(TFileSystemTest)
             AtomicGet(counters->GetCounter("InProgress")->GetAtomic()));
     }
 
+    Y_UNIT_TEST(ShouldDrainHandleOpsQueueBackToBack)
+    {
+        // AsyncHandleOperationDrainPeriod is 0, so a non-empty queue is drained
+        // back-to-back: every entry is rescheduled with zero delay.
+        // With a frozen clock, running only the tasks due "now" fires just the
+        // zero-delay tasks, so the whole queue must drain without advancing
+        // time. A non-zero period would leave every processing task in the
+        // future and this drain would never complete.
+        constexpr ui32 requestCount = 3;
+
+        NProto::TFileStoreFeatures features;
+        features.SetAsyncDestroyHandleEnabled(true);
+        features.SetAsyncHandleOperationIdlePeriod(50);
+        features.SetAsyncHandleOperationDrainPeriod(0);
+
+        auto timer = std::make_shared<TTestTimer>();
+        auto scheduler = std::make_shared<TTestScheduler>(timer->Now());
+        TBootstrap bootstrap(timer, scheduler, features);
+
+        std::atomic_uint handlerCalled = 0;
+        bootstrap.Service->SetHandlerDestroyHandle(
+            [&](auto, auto)
+            {
+                ++handlerCalled;
+                return MakeFuture(NProto::TDestroyHandleResponse{});
+            });
+
+        auto inProgress =
+            bootstrap.Counters->FindSubgroup("component", "fs_ut")
+                ->FindSubgroup("request", "DestroyHandle")
+                ->GetCounter("InProgress");
+
+        bootstrap.Start();
+        Y_DEFER {
+            bootstrap.Stop();
+        };
+
+        for (ui32 i = 0; i < requestCount; ++i) {
+            auto future = bootstrap.Fuse->SendRequest<TReleaseRequest>(
+                10 + i,
+                2 + i,
+                O_RDONLY);
+            UNIT_ASSERT_NO_EXCEPTION(future.GetValue(WaitTimeout));
+        }
+
+        UNIT_ASSERT(WaitForCondition(
+            WaitTimeout,
+            [&]
+            {
+                scheduler->RunAllScheduledTasksUntilNow();
+                return handlerCalled.load() == requestCount
+                    && AtomicGet(inProgress->GetAtomic()) == 0;
+            }));
+    }
+
+    Y_UNIT_TEST(ShouldUseIdlePeriodWhenHandleOpsQueueEmpty)
+    {
+        NProto::TFileStoreFeatures features;
+        features.SetAsyncDestroyHandleEnabled(true);
+        features.SetAsyncHandleOperationIdlePeriod(50);
+        features.SetAsyncHandleOperationDrainPeriod(0);
+        const auto idlePeriod = TDuration::MilliSeconds(
+            features.GetAsyncHandleOperationIdlePeriod());
+
+        auto timer = std::make_shared<TTestTimer>();
+        auto scheduler = std::make_shared<TTestScheduler>(timer->Now());
+        TBootstrap bootstrap(timer, scheduler, features);
+
+        std::atomic_uint handlerCalled = 0;
+        bootstrap.Service->SetHandlerDestroyHandle(
+            [&](auto, auto)
+            {
+                ++handlerCalled;
+                return MakeFuture(NProto::TDestroyHandleResponse{});
+            });
+
+        bootstrap.Start();
+        Y_DEFER {
+            bootstrap.Stop();
+        };
+
+        scheduler->RunAllScheduledTasksUntilNow();
+
+        auto future = bootstrap.Fuse->SendRequest<TReleaseRequest>(
+            10,
+            2,
+            O_RDONLY);
+        UNIT_ASSERT_NO_EXCEPTION(future.GetValue(WaitTimeout));
+
+        // The only scheduled poll is idlePeriod in the future, so at the
+        // current (frozen) time it does not fire and the entry stays pending.
+        scheduler->RunAllScheduledTasksUntilNow();
+        UNIT_ASSERT_VALUES_EQUAL(0U, handlerCalled.load());
+
+        // Once the backoff elapses the poll fires and picks up the entry.
+        timer->AdvanceTime(idlePeriod);
+        scheduler->AdvanceTime(idlePeriod);
+        UNIT_ASSERT(WaitForCondition(
+            WaitTimeout,
+            [&]
+            {
+                scheduler->RunAllScheduledTasksUntilNow();
+                return handlerCalled.load() == 1;
+            }));
+    }
+
     Y_UNIT_TEST(ShouldProcessReadOnlyDestroyHandleRequestsAsynchronously)
     {
         NProto::TFileStoreFeatures features;

cloud/filestore/libs/vfs_fuse/loop.cpp

@@ -1299,8 +1299,14 @@ class TFileSystemLoop final
             features.GetAsyncDestroyReadOnlyHandleEnabled());
         config.SetAsyncCreateHandleEnabled(
             features.GetAsyncCreateHandleEnabled());
-        config.SetAsyncHandleOperationPeriod(
-            features.GetAsyncHandleOperationPeriod());
+        // If the idle period is not published by the server, fall back to the
+        // default period.
+        if (features.GetAsyncHandleOperationIdlePeriod()) {
+            config.SetAsyncHandleOperationIdlePeriod(
+                features.GetAsyncHandleOperationIdlePeriod());
+        }
+        config.SetAsyncHandleOperationDrainPeriod(
+            features.GetAsyncHandleOperationDrainPeriod());
 
         config.SetDirectIoEnabled(features.GetDirectIoEnabled());
         config.SetDirectIoAlign(features.GetDirectIoAlign());