issue-1751: [Filestore] WriteBackCache - refactor/rewrite main state (#5292)

Author: e673

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

@@ -0,0 +1,80 @@
+#include "node_flush_state.h"
+
+#include <cloud/storage/core/libs/common/error.h>
+
+#include <util/generic/algorithm.h>
+
+namespace NCloud::NFileStore::NFuse::NWriteBackCache {
+
+////////////////////////////////////////////////////////////////////////////////
+
+TNodeFlushState::TNodeFlushState(
+    ui64 nodeId,
+    TVector<std::shared_ptr<NProto::TWriteDataRequest>> requests,
+    ui64 affectedUnflushedRequestCount)
+    : NodeId(nodeId)
+    , AffectedUnflushedRequestCount(affectedUnflushedRequestCount)
+{
+    Y_ABORT_UNLESS(!requests.empty());
+    Y_ABORT_UNLESS(affectedUnflushedRequestCount > 0);
+
+    for (auto& request: requests) {
+        WriteDataRequests.emplace_back(std::move(request));
+    }
+}
+
+TVector<std::shared_ptr<NProto::TWriteDataRequest>>
+TNodeFlushState::BeginFlush()
+{
+    Y_ABORT_UNLESS(InFlightWriteDataRequestCount.load() == 0);
+    Y_ABORT_UNLESS(!WriteDataRequests.empty(), "Nothing to flush");
+
+    TVector<std::shared_ptr<NProto::TWriteDataRequest>> res(
+        Reserve(WriteDataRequests.size()));
+
+    for (const auto& it: WriteDataRequests) {
+        res.push_back(it.Request);
+    }
+
+    InFlightWriteDataRequestCount.store(WriteDataRequests.size());
+
+    return res;
+}
+
+EWriteDataRequestCompletedAction TNodeFlushState::OnWriteDataRequestCompleted(
+    size_t index,
+    const NProto::TWriteDataResponse& response)
+{
+    Y_ABORT_UNLESS(index < WriteDataRequests.size());
+
+    WriteDataRequests[index].Error = response.GetError();
+
+    auto prev =
+        InFlightWriteDataRequestCount.fetch_sub(1, std::memory_order_release);
+    Y_ABORT_UNLESS(prev > 0);
+
+    if (prev > 1) {
+        return EWriteDataRequestCompletedAction::ContinueExecution;
+    }
+
+    // The thread that has made the final decrement should process the results
+    // of other threads
+    auto cur = InFlightWriteDataRequestCount.load(std::memory_order_acquire);
+    Y_ABORT_UNLESS(cur == 0);
+
+    return EWriteDataRequestCompletedAction::CollectFlushResult;
+}
+
+NCloud::NProto::TError TNodeFlushState::CollectFlushResult()
+{
+    Y_ABORT_UNLESS(InFlightWriteDataRequestCount.load() == 0);
+
+    EraseIf(
+        WriteDataRequests,
+        [](const auto& it) { return !HasError(it.Error); });
+
+    return WriteDataRequests.empty() ? NCloud::NProto::TError()
+                                     : WriteDataRequests.front().Error;
+}
+
+}   // namespace NCloud::NFileStore::NFuse::NWriteBackCache

cloud/filestore/libs/vfs_fuse/write_back_cache/node_flush_state.h

@@ -0,0 +1,89 @@
+#pragma once
+
+#include <cloud/filestore/public/api/protos/data.pb.h>
+
+#include <util/generic/vector.h>
+
+#include <atomic>
+
+namespace NCloud::NFileStore::NFuse::NWriteBackCache {
+
+////////////////////////////////////////////////////////////////////////////////
+
+struct TWriteDataRequestWithResponse
+{
+    std::shared_ptr<NProto::TWriteDataRequest> Request;
+    NCloud::NProto::TError Error;
+
+    explicit TWriteDataRequestWithResponse(
+        std::shared_ptr<NProto::TWriteDataRequest> request)
+        : Request(std::move(request))
+    {}
+};
+
+////////////////////////////////////////////////////////////////////////////////
+
+// The action that the caller should take after calling
+// TNodeFlushState::OnWriteDataRequestCompleted
+enum class EWriteDataRequestCompletedAction
+{
+    // The caller should continue executing other requests
+    ContinueExecution,
+
+    // The caller was the last executing thread and should collect flush result
+    CollectFlushResult
+};
+
+////////////////////////////////////////////////////////////////////////////////
+
+class TNodeFlushState
+{
+private:
+    const ui64 NodeId;
+
+    // WriteData requests generated during Flush that are executed in parallel
+    TVector<TWriteDataRequestWithResponse> WriteDataRequests;
+
+    // The amount of unflushed requests that will become flushed when all
+    // requests from |WriteDataRequests| are completed
+    ui64 AffectedUnflushedRequestCount = 0;
+
+    // The number of remaining executing requests from |WriteDataRequests|
+    std::atomic<ui64> InFlightWriteDataRequestCount = 0;
+
+public:
+    TNodeFlushState(
+        ui64 nodeId,
+        TVector<std::shared_ptr<NProto::TWriteDataRequest>> requests,
+        ui64 affectedUnflushedRequestCount);
+
+    ui64 GetNodeId() const
+    {
+        return NodeId;
+    }
+
+    ui64 GetAffectedUnflushedRequestCount() const
+    {
+        return AffectedUnflushedRequestCount;
+    }
+
+    size_t GetWriteDataRequestCount() const
+    {
+        return WriteDataRequests.size();
+    }
+
+    // The caller should execute the requests in parallel and call
+    // OnWriteDataRequestCompleted for each response with the index
+    // corresponding to the request in the returned vector
+    TVector<std::shared_ptr<NProto::TWriteDataRequest>> BeginFlush();
+
+    // The method is thread-safe
+    EWriteDataRequestCompletedAction OnWriteDataRequestCompleted(
+        size_t index,
+        const NProto::TWriteDataResponse& response);
+
+    // When error is returned, the caller should retry flush
+    NCloud::NProto::TError CollectFlushResult();
+};
+
+}   // namespace NCloud::NFileStore::NFuse::NWriteBackCache

cloud/filestore/libs/vfs_fuse/write_back_cache/node_state.h

@@ -13,13 +13,16 @@ namespace NCloud::NFileStore::NFuse::NWriteBackCache {
 
 ////////////////////////////////////////////////////////////////////////////////
 
-struct TNodeFlushState
+enum class ENodeFlushStatus
 {
-    TVector<std::shared_ptr<NProto::TWriteDataRequest>> WriteRequests;
-    TVector<std::shared_ptr<NProto::TWriteDataRequest>> FailedWriteRequests;
-    size_t AffectedWriteDataEntriesCount = 0;
-    size_t InFlightWriteRequestsCount = 0;
-    bool Executing = false;
+    // Node has no unflushed requests
+    NothingToFlush,
+
+    // Node has unflushed requests but flush has not been requested yet
+    ReadyToFlush,
+
+    // Flush has been requested for a node
+    FlushRequested
 };
 
 ////////////////////////////////////////////////////////////////////////////////
@@ -41,58 +44,53 @@ struct TFlushRequest
 
 struct TNodeState
 {
-    const ui64 NodeId;
-
     // Holds pending, unflushed and flushed requests
     // Tracks cached data parts
     TNodeCache Cache;
 
     // Prevent from concurrent read and write requests with overlapping ranges
     TReadWriteRangeLock RangeLock;
 
-    TNodeFlushState FlushState;
+    ENodeFlushStatus FlushStatus = ENodeFlushStatus::NothingToFlush;
 
     // Flush requests are fulfilled when there are no pending or unflushed
     // requests with SequenceId less or equal than |TFlushRequest::SequenceId|.
     // Flush requests are stored in chronological order: SequenceId values are
     // strictly increasing so newer flush requests have larger SequenceId.
     TDeque<TFlushRequest> FlushRequests;
 
-    // All entries with RequestId <= |AutomaticFlushRequestId| are to be flushed
-    ui64 AutomaticFlushRequestId = 0;
-
     // Cached data extends the node size but until the data is flushed,
     // the changes are not visible to the tablet. FileSystem requests that
     // return node attributes or rely on it (GetAttr, Lookup, Read, ReadDir)
     // should have the node size adjusted to this value.
     ui64 CachedNodeSize = 0;
 
-    explicit TNodeState(ui64 nodeId)
-        : NodeId(nodeId)
-    {}
-
     bool CanBeDeleted() const
     {
         if (Cache.Empty() && RangeLock.Empty()) {
-            Y_ABORT_UNLESS(!FlushState.Executing);
+            Y_ABORT_UNLESS(FlushRequests.empty());
+            Y_ABORT_UNLESS(FlushStatus == ENodeFlushStatus::NothingToFlush);
             return true;
         }
         return false;
     }
 
-    bool ShouldFlush(ui64 maxFlushAllRequestId) const
+    ENodeFlushStatus GetExpectedFlushStatus(ui64 flushAllSequenceId) const
     {
+        if (FlushStatus == ENodeFlushStatus::FlushRequested) {
+            // Once Flush has been scheduled, the status can be changed only in
+            // FlushSucceeded and FlushFailed calls
+            return ENodeFlushStatus::FlushRequested;
+        }
         if (!Cache.HasUnflushedRequests()) {
-            return false;
+            return ENodeFlushStatus::NothingToFlush;
         }
-
-        if (!FlushRequests.empty()) {
-            return true;
+        if (!FlushRequests.empty() ||
+            Cache.GetMinUnflushedSequenceId() <= flushAllSequenceId)
+        {
+            return ENodeFlushStatus::FlushRequested;
         }
-
-        const ui64 minRequestId = Cache.GetMinUnflushedSequenceId();
-        return minRequestId <= maxFlushAllRequestId ||
-               minRequestId <= AutomaticFlushRequestId;
+        return ENodeFlushStatus::ReadyToFlush;
     }
 };
 

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

@@ -10,7 +10,7 @@ TNodeStateHolder::TNodeStateHolder(IWriteBackCacheStatsPtr stats)
 
 TNodeState& TNodeStateHolder::GetOrCreateNodeState(ui64 nodeId)
 {
-    auto [it, inserted] = NodeStates.try_emplace(nodeId, TNodeState(nodeId));
+    auto [it, inserted] = NodeStates.try_emplace(nodeId);
     if (inserted) {
         Stats->IncrementNodeCount();
     }

cloud/filestore/libs/vfs_fuse/write_back_cache/node_state_holder.h

@@ -53,11 +53,11 @@ class TNodeStateHolder
     // Return a pointer to the node state for nodeId or nullptr if it does not
     // exist.
     //
-    // If includeDeleted is false (default), logically deleted states are
-    // treated as absent and nullptr may be returned. If includeDeleted is
-    // true, the function will return states that were deleted but are still
-    // retained due to active pins.
-    TNodeState* GetNodeState(ui64 nodeId, bool includeDeleted = false);
+    // If includeDeleted is false, logically deleted states are treated as
+    // absent and nullptr may be returned. If includeDeleted is true, the
+    // function will return states that were deleted but are still retained due
+    // to active pins.
+    TNodeState* GetNodeState(ui64 nodeId, bool includeDeleted);
 
     const TNodeState* GetNodeState(
         ui64 nodeId,

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

@@ -0,0 +1,94 @@
+#include "queued_operations.h"
+
+#include <variant>
+
+namespace NCloud::NFileStore::NFuse::NWriteBackCache {
+
+namespace {
+
+////////////////////////////////////////////////////////////////////////////////
+
+struct TWriteDataPromiseCompletedEvent
+{
+    NThreading::TPromise<NProto::TWriteDataResponse> Promise;
+
+    void Invoke()
+    {
+        Promise.SetValue({});
+    }
+};
+
+struct TFlushPromiseCompletedEvent
+{
+    NThreading::TPromise<void> Promise;
+
+    void Invoke()
+    {
+        Promise.SetValue();
+    }
+};
+
+struct TScheduleFlushEvent
+{
+    IQueuedOperationsProcessor& Processor;
+    ui64 NodeId;
+
+    void Invoke()
+    {
+        Processor.ScheduleFlushNode(NodeId);
+    }
+};
+
+using TEventVariant = std::variant<
+    TWriteDataPromiseCompletedEvent,
+    TFlushPromiseCompletedEvent,
+    TScheduleFlushEvent>;
+
+}   // namespace
+
+////////////////////////////////////////////////////////////////////////////////
+
+struct TQueuedOperations::TEvent: public TEventVariant
+{
+    using TEventVariant::TEventVariant;
+};
+
+////////////////////////////////////////////////////////////////////////////////
+
+TQueuedOperations::TQueuedOperations(IQueuedOperationsProcessor& processor)
+    : Processor(processor)
+{}
+
+TQueuedOperations::~TQueuedOperations() = default;
+
+void TQueuedOperations::Acquire()
+{
+    Lock.Acquire();
+}
+
+void TQueuedOperations::Release()
+{
+    auto events = std::exchange(Events, {});
+    Lock.Release();
+    for (auto& event: events) {
+        std::visit([](auto& ev) { ev.Invoke(); }, event);
+    }
+}
+
+void TQueuedOperations::ScheduleFlushNode(ui64 nodeId)
+{
+    Events.push_back(TScheduleFlushEvent{Processor, nodeId});
+}
+
+void TQueuedOperations::CompleteWriteDataPromise(
+    NThreading::TPromise<NProto::TWriteDataResponse> promise)
+{
+    Events.push_back(TWriteDataPromiseCompletedEvent{std::move(promise)});
+}
+
+void TQueuedOperations::CompleteFlushPromise(NThreading::TPromise<void> promise)
+{
+    Events.push_back(TFlushPromiseCompletedEvent{std::move(promise)});
+}
+
+}   // namespace NCloud::NFileStore::NFuse::NWriteBackCache