issue-1751: [Filestore] Simplify WriteBackCache logic (deduplicate code for handling pending requests) (#4561)

Author: e673

cloud/filestore/libs/vfs_fuse/write_back_cache/write_back_cache.cpp

@@ -116,10 +116,6 @@ struct TWriteBackCache::TNodeState
     // Efficient calculation of TWriteDataEntryParts from CachedEntries
     TWriteDataEntryIntervalMap CachedEntryIntervalMap;
 
-    // Count entries in TWriteBackCache::TImpl::PendingEntries
-    // with status Pending filtered by nodeId
-    size_t PendingEntriesCount = 0;
-
     // Prevent from concurrent read and write requests with overlapping ranges
     TReadWriteRangeLock RangeLock;
 
@@ -129,10 +125,10 @@ struct TWriteBackCache::TNodeState
         : NodeId(nodeId)
     {}
 
-    bool Empty() const
+    bool CanBeDeleted() const
     {
-        return CachedEntries.empty() && PendingEntriesCount == 0 &&
-               RangeLock.Empty() && !FlushState.Executing;
+        return CachedEntries.empty() && RangeLock.Empty() &&
+               !FlushState.Executing;
     }
 
     bool ShouldFlush() const
@@ -153,16 +149,14 @@ struct TWriteBackCache::TPendingOperations
     bool Executing = false;
 
     // The flag is set when an element is popped from
-    // |TImpl::CachedEntriesPersistentQueue| and free space is increased.
+    // |TImpl::CachedEntriesPersistentQueue| and free space is increased or
+    // an element is pushed to empty |TImpl::CachedEntriesPersistentQueue|.
     // We should try to push pending entries to the persistent queue.
     bool ShouldProcessPendingEntries = false;
 
     // Pending ReadData requests that have acquired |TNodeState::RangeLock|
     TVector<TPendingReadDataRequest> ReadData;
 
-    // Pending WriteData requests that have acquired |TNodeState::RangeLock|
-    TVector<std::unique_ptr<TWriteDataEntry>> WriteData;
-
     // Flush operations that have been scheduled but not yet started
     TVector<TNodeState*> Flush;
 
@@ -174,7 +168,7 @@ struct TWriteBackCache::TPendingOperations
 
     bool Empty() const
     {
-        return ReadData.empty() && WriteData.empty() && Flush.empty() &&
+        return ReadData.empty() && Flush.empty() &&
                WriteDataCompleted.empty() && FlushCompleted.empty();
     }
 };
@@ -456,7 +450,7 @@ class TWriteBackCache::TImpl final
                 auto guard = Guard(self->Lock);
                 auto* nodeState = self->GetNodeState(nodeId);
                 nodeState->RangeLock.UnlockRead(offset, end);
-                self->DeleteNodeStateIfEmpty(nodeState);
+                self->DeleteNodeStateIfNeeded(nodeState);
                 self->ExecutePendingOperations(guard);
             }
         };
@@ -529,7 +523,7 @@ class TWriteBackCache::TImpl final
                 auto guard = Guard(self->Lock);
                 auto* nodeState = self->GetNodeState(nodeId);
                 nodeState->RangeLock.UnlockWrite(offset, end);
-                self->DeleteNodeStateIfEmpty(nodeState);
+                self->DeleteNodeStateIfNeeded(nodeState);
                 self->ExecutePendingOperations(guard);
             }
         };
@@ -548,7 +542,10 @@ class TWriteBackCache::TImpl final
             // TImpl is alive
             Y_ABORT_UNLESS(self);
             if (self) {
-                self->PendingOperations.WriteData.push_back(
+                if (self->PendingEntries.empty()) {
+                    self->PendingOperations.ShouldProcessPendingEntries = true;
+                }
+                self->PendingEntries.push_back(
                     std::unique_ptr<TWriteDataEntry>(entry));
             }
         };
@@ -718,9 +715,9 @@ class TWriteBackCache::TImpl final
         return ptr.get();
     }
 
-    void DeleteNodeStateIfEmpty(TNodeState* nodeState)
+    void DeleteNodeStateIfNeeded(TNodeState* nodeState)
     {
-        if (nodeState != nullptr && nodeState->Empty()) {
+        if (nodeState != nullptr && nodeState->CanBeDeleted()) {
             auto erased = NodeStates.erase(nodeState->NodeId);
             Y_DEBUG_ABORT_UNLESS(erased);
             Stats->DecrementNodeCount();
@@ -751,13 +748,11 @@ class TWriteBackCache::TImpl final
             }
 
             TVector<TPendingReadDataRequest> readData;
-            TVector<std::unique_ptr<TWriteDataEntry>> writeData;
             TVector<TNodeState*> flush;
             TVector<TPromise<NProto::TWriteDataResponse>> writeDataCompleted;
             TVector<TPromise<void>> flushCompleted;
 
             swap(readData, PendingOperations.ReadData);
-            swap(writeData, PendingOperations.WriteData);
             swap(flush, PendingOperations.Flush);
             swap(writeDataCompleted, PendingOperations.WriteDataCompleted);
             swap(flushCompleted, PendingOperations.FlushCompleted);
@@ -770,10 +765,6 @@ class TWriteBackCache::TImpl final
                 StartPendingReadDataRequest(std::move(request));
             }
 
-            for (auto& entry: writeData) {
-                StartPendingWriteDataRequest(std::move(entry));
-            }
-
             for (auto* nodeState: flush) {
                 StartFlush(nodeState);
             }
@@ -817,8 +808,6 @@ class TWriteBackCache::TImpl final
             auto* nodeState = GetNodeState(entry->GetNodeId());
             AddCachedEntry(nodeState, std::move(entry));
 
-            Y_ABORT_UNLESS(nodeState->PendingEntriesCount > 0);
-            nodeState->PendingEntriesCount--;
             PendingEntries.pop_front();
         }
 
@@ -1024,44 +1013,6 @@ class TWriteBackCache::TImpl final
         state.Promise.SetValue(std::move(response));
     }
 
-    void StartPendingWriteDataRequest(std::unique_ptr<TWriteDataEntry> entry)
-    {
-        auto serializedSize = entry->GetSerializedSize();
-        auto guard = Guard(Lock);
-
-        auto* nodeState = GetNodeState(entry->GetNodeId());
-
-        if (nodeState->PendingEntriesCount > 0)
-        {
-            nodeState->PendingEntriesCount++;
-            PendingEntries.push_back(std::move(entry));
-            return;
-        }
-
-        char* allocationPtr = nullptr;
-        bool allocated = CachedEntriesPersistentQueue.AllocateBack(
-            serializedSize,
-            &allocationPtr);
-
-        if (allocated) {
-            Y_ABORT_UNLESS(allocationPtr != nullptr);
-
-            entry->SerializeAndMoveRequestBuffer(
-                allocationPtr,
-                PendingOperations,
-                this);
-
-            CachedEntriesPersistentQueue.CommitAllocation(allocationPtr);
-            AddCachedEntry(nodeState, std::move(entry));
-        } else {
-            nodeState->PendingEntriesCount++;
-            PendingEntries.push_back(std::move(entry));
-            RequestFlushAll();
-        }
-
-        UpdatePersistentQueueStats();
-    }
-
     // |nodeState| becomes unusable if the function returns false
     void PrepareFlush(TNodeState* nodeState)
     {
@@ -1218,7 +1169,7 @@ class TWriteBackCache::TImpl final
             PendingOperations.Flush.push_back(nodeState);
         } else {
             nodeState->FlushState.Executing = false;
-            DeleteNodeStateIfEmpty(nodeState);
+            DeleteNodeStateIfNeeded(nodeState);
         }
 
         ExecutePendingOperations(guard);