From 8ceaf58309fd3d5fa60d41ae0478379485384905 Mon Sep 17 00:00:00 2001 From: Evgeny Budilovsky Date: Thu, 23 Jul 2026 09:31:06 +0000 Subject: [PATCH 1/2] [Filestore] issue-5432: add basic async open support (vfs_fuse) --- cloud/filestore/config/filesystem.proto | 3 + cloud/filestore/config/storage.proto | 3 + .../libs/diagnostics/profile_log_events.cpp | 12 + .../diagnostics/profile_log_events_ut.cpp | 30 + cloud/filestore/libs/service/auth_scheme.cpp | 1 + cloud/filestore/libs/service/request.h | 1 + cloud/filestore/libs/service_local/fs.h | 1 + .../filestore/libs/service_local/fs_data.cpp | 8 + cloud/filestore/libs/service_null/service.cpp | 10 + cloud/filestore/libs/vfs_fuse/config.cpp | 1 + cloud/filestore/libs/vfs_fuse/config.h | 1 + cloud/filestore/libs/vfs_fuse/fs_impl.cpp | 114 ++- cloud/filestore/libs/vfs_fuse/fs_impl.h | 33 + .../filestore/libs/vfs_fuse/fs_impl_data.cpp | 286 +++++- cloud/filestore/libs/vfs_fuse/fs_ut.cpp | 934 +++++++++++++++++- .../libs/vfs_fuse/handle_ops_queue.cpp | 28 + .../libs/vfs_fuse/handle_ops_queue.h | 6 +- cloud/filestore/libs/vfs_fuse/loop.cpp | 8 +- .../libs/vfs_fuse/protos/queue_entry.proto | 9 + cloud/filestore/public/api/grpc/service.proto | 7 + cloud/filestore/public/api/protos/data.proto | 38 + cloud/filestore/public/api/protos/fs.proto | 1 + .../analytics/libs/event-log/dump_ut.cpp | 31 +- 23 files changed, 1509 insertions(+), 57 deletions(-) diff --git a/cloud/filestore/config/filesystem.proto b/cloud/filestore/config/filesystem.proto index 563386b63e6..bd647151a70 100644 --- a/cloud/filestore/config/filesystem.proto +++ b/cloud/filestore/config/filesystem.proto @@ -124,4 +124,7 @@ message TFileSystemConfig // Async processing of destroy handle requests only for read only handles. optional bool AsyncDestroyReadOnlyHandleEnabled = 36; + + // Async processing of read-only create handle requests. + optional bool AsyncCreateHandleEnabled = 37; } diff --git a/cloud/filestore/config/storage.proto b/cloud/filestore/config/storage.proto index fd6c086b5b4..6cb02ad728e 100644 --- a/cloud/filestore/config/storage.proto +++ b/cloud/filestore/config/storage.proto @@ -851,4 +851,7 @@ message TStorageConfig // Enables injecting of errors in TIndexTabletDatabase. Must be only used in tests! optional double FakeTxPageFaultsProbability = 527; + + // Async processing of read-only create handle requests. + optional bool AsyncCreateHandleEnabled = 528; } diff --git a/cloud/filestore/libs/diagnostics/profile_log_events.cpp b/cloud/filestore/libs/diagnostics/profile_log_events.cpp index c3e478cf17e..5ebe6bea994 100644 --- a/cloud/filestore/libs/diagnostics/profile_log_events.cpp +++ b/cloud/filestore/libs/diagnostics/profile_log_events.cpp @@ -255,6 +255,17 @@ void InitProfileLogRequestInfo( nodeInfo->SetHandle(request.GetHandle()); } +template <> +void InitProfileLogRequestInfo( + NProto::TProfileLogRequestInfo& profileLogRequest, + const NProto::TConfirmCreateHandleRequest& request) +{ + auto* nodeInfo = profileLogRequest.MutableNodeInfo(); + nodeInfo->SetNodeId(request.GetNodeId()); + nodeInfo->SetHandle(request.GetHandle()); + nodeInfo->SetFlags(request.GetFlags()); +} + template <> void InitProfileLogRequestInfo( NProto::TProfileLogRequestInfo& profileLogRequest, @@ -666,6 +677,7 @@ void UpdateRangeNodeIds( IMPLEMENT_DEFAULT_METHOD(AccessNode, NProto) IMPLEMENT_DEFAULT_METHOD(ReadLink, NProto) IMPLEMENT_DEFAULT_METHOD(RemoveNodeXAttr, NProto) + IMPLEMENT_DEFAULT_METHOD(ConfirmCreateHandle, NProto) IMPLEMENT_DEFAULT_METHOD(DestroyHandle, NProto) IMPLEMENT_DEFAULT_METHOD(AcquireLock, NProto) IMPLEMENT_DEFAULT_METHOD(ReleaseLock, NProto) diff --git a/cloud/filestore/libs/diagnostics/profile_log_events_ut.cpp b/cloud/filestore/libs/diagnostics/profile_log_events_ut.cpp index cec853e680f..77f0121373c 100644 --- a/cloud/filestore/libs/diagnostics/profile_log_events_ut.cpp +++ b/cloud/filestore/libs/diagnostics/profile_log_events_ut.cpp @@ -110,6 +110,36 @@ Y_UNIT_TEST_SUITE(TProfileLogEventsTest) UNIT_ASSERT(!nodeInfo.HasSize()); } + Y_UNIT_TEST(ShouldConfirmCreateHandleRequestInitializeFieldsCorrectly) + { + const auto nodeId = 12; + const auto handle = 34; + const auto flags = 56; + + NProto::TConfirmCreateHandleRequest req; + req.SetNodeId(nodeId); + req.SetHandle(handle); + req.SetFlags(flags); + + NProto::TProfileLogRequestInfo profileLogRequest; + InitProfileLogRequestInfo(profileLogRequest, req); + + UNIT_ASSERT_VALUES_EQUAL(0, profileLogRequest.RangesSize()); + UNIT_ASSERT(profileLogRequest.HasNodeInfo()); + UNIT_ASSERT(!profileLogRequest.HasLockInfo()); + + const auto& nodeInfo = profileLogRequest.GetNodeInfo(); + UNIT_ASSERT(!nodeInfo.HasParentNodeId()); + UNIT_ASSERT(!nodeInfo.HasNodeName()); + UNIT_ASSERT(!nodeInfo.HasNewParentNodeId()); + UNIT_ASSERT(!nodeInfo.HasNewNodeName()); + UNIT_ASSERT_VALUES_EQUAL(flags, nodeInfo.GetFlags()); + UNIT_ASSERT(!nodeInfo.HasMode()); + UNIT_ASSERT_VALUES_EQUAL(nodeId, nodeInfo.GetNodeId()); + UNIT_ASSERT_VALUES_EQUAL(handle, nodeInfo.GetHandle()); + UNIT_ASSERT(!nodeInfo.HasSize()); + } + Y_UNIT_TEST(ShouldReadDataRequestInitializeFieldsCorrectly) { const auto nodeId = 12; diff --git a/cloud/filestore/libs/service/auth_scheme.cpp b/cloud/filestore/libs/service/auth_scheme.cpp index 58684ce01d6..cea6dcbe587 100644 --- a/cloud/filestore/libs/service/auth_scheme.cpp +++ b/cloud/filestore/libs/service/auth_scheme.cpp @@ -39,6 +39,7 @@ TPermissionList GetRequestPermissions(EFileStoreRequest requestType) case EFileStoreRequest::SetNodeXAttr: case EFileStoreRequest::RemoveNodeXAttr: case EFileStoreRequest::CreateHandle: + case EFileStoreRequest::ConfirmCreateHandle: case EFileStoreRequest::DestroyHandle: case EFileStoreRequest::WriteData: case EFileStoreRequest::AllocateData: diff --git a/cloud/filestore/libs/service/request.h b/cloud/filestore/libs/service/request.h index d6f77dc3202..2a99def8527 100644 --- a/cloud/filestore/libs/service/request.h +++ b/cloud/filestore/libs/service/request.h @@ -72,6 +72,7 @@ namespace NCloud::NFileStore { xxx(ReadData, __VA_ARGS__) \ xxx(WriteData, __VA_ARGS__) \ xxx(AllocateData, __VA_ARGS__) \ + xxx(ConfirmCreateHandle, __VA_ARGS__) \ // FILESTORE_DATA_METHODS #define FILESTORE_LOCAL_DATA_METHODS(xxx, ...) \ diff --git a/cloud/filestore/libs/service_local/fs.h b/cloud/filestore/libs/service_local/fs.h index ca1955de801..05af6215eb1 100644 --- a/cloud/filestore/libs/service_local/fs.h +++ b/cloud/filestore/libs/service_local/fs.h @@ -52,6 +52,7 @@ namespace NCloud::NFileStore { xxx(RemoveNodeXAttr, __VA_ARGS__) \ \ xxx(CreateHandle, __VA_ARGS__) \ + xxx(ConfirmCreateHandle, __VA_ARGS__) \ xxx(DestroyHandle, __VA_ARGS__) \ \ xxx(AcquireLock, __VA_ARGS__) \ diff --git a/cloud/filestore/libs/service_local/fs_data.cpp b/cloud/filestore/libs/service_local/fs_data.cpp index 309fcdc7178..02577c53033 100644 --- a/cloud/filestore/libs/service_local/fs_data.cpp +++ b/cloud/filestore/libs/service_local/fs_data.cpp @@ -91,6 +91,14 @@ NProto::TCreateHandleResponse TLocalFileSystem::CreateHandle( return response; } +NProto::TConfirmCreateHandleResponse TLocalFileSystem::ConfirmCreateHandle( + const NProto::TConfirmCreateHandleRequest& request) +{ + Y_UNUSED(request); + + return {}; +} + NProto::TDestroyHandleResponse TLocalFileSystem::DestroyHandle( const NProto::TDestroyHandleRequest& request) { diff --git a/cloud/filestore/libs/service_null/service.cpp b/cloud/filestore/libs/service_null/service.cpp index 3a740392e2d..d0b36f72568 100644 --- a/cloud/filestore/libs/service_null/service.cpp +++ b/cloud/filestore/libs/service_null/service.cpp @@ -190,6 +190,16 @@ struct TNullFileStore final return MakeFuture(response); } + TFuture ConfirmCreateHandle( + TCallContextPtr callContext, + std::shared_ptr request) override + { + Y_UNUSED(callContext); + Y_UNUSED(request); + + return MakeFuture(NProto::TConfirmCreateHandleResponse{}); + } + TFuture ReadData( TCallContextPtr callContext, std::shared_ptr request) override diff --git a/cloud/filestore/libs/vfs_fuse/config.cpp b/cloud/filestore/libs/vfs_fuse/config.cpp index b032ad96233..dee9c0d6e10 100644 --- a/cloud/filestore/libs/vfs_fuse/config.cpp +++ b/cloud/filestore/libs/vfs_fuse/config.cpp @@ -31,6 +31,7 @@ namespace { \ xxx(AsyncDestroyHandleEnabled, bool, false )\ xxx(AsyncDestroyReadOnlyHandleEnabled, bool, false )\ + xxx(AsyncCreateHandleEnabled, bool, false )\ xxx(AsyncHandleOperationPeriod, TDuration, TDuration::MilliSeconds(50) )\ \ xxx(DirectIoEnabled, bool, false )\ diff --git a/cloud/filestore/libs/vfs_fuse/config.h b/cloud/filestore/libs/vfs_fuse/config.h index 1086bab6ecd..0ac7660d716 100644 --- a/cloud/filestore/libs/vfs_fuse/config.h +++ b/cloud/filestore/libs/vfs_fuse/config.h @@ -40,6 +40,7 @@ struct TFileSystemConfig bool GetAsyncDestroyHandleEnabled() const; bool GetAsyncDestroyReadOnlyHandleEnabled() const; + bool GetAsyncCreateHandleEnabled() const; TDuration GetAsyncHandleOperationPeriod() const; bool GetDirectIoEnabled() const; diff --git a/cloud/filestore/libs/vfs_fuse/fs_impl.cpp b/cloud/filestore/libs/vfs_fuse/fs_impl.cpp index f2c20e4bd02..0abf70b2ff2 100644 --- a/cloud/filestore/libs/vfs_fuse/fs_impl.cpp +++ b/cloud/filestore/libs/vfs_fuse/fs_impl.cpp @@ -87,7 +87,7 @@ TFileSystem::~TFileSystem() void TFileSystem::Init() { - STORAGE_INFO("scheduling destroy handle queue processing"); + STORAGE_INFO("scheduling handle ops queue processing"); ScheduleProcessHandleOpsQueue(); } @@ -376,45 +376,82 @@ void TFileSystem::CompleteAsyncDestroyHandle( const auto& error = response.GetError(); RequestStats->RequestCompleted(callContext, error); - // If destroy request failed, we need to retry it. - // Otherwise, remove it from queue. if (HasError(error)) { STORAGE_ERROR( "DestroyHandle request failed: " << "filesystem " << Config->GetFileSystemId() << " error: " << FormatError(error)); - if (GetErrorKind(error) != EErrorKind::ErrorRetriable) { - ReportAsyncDestroyHandleFailed(); - with_lock (HandleOpsQueueLock) { - HandleOpsQueue->PopFront(); - } + if (GetErrorKind(error) == EErrorKind::ErrorRetriable) { + ScheduleProcessHandleOpsQueue(); + return; } - } else { - with_lock (HandleOpsQueueLock) { - HandleOpsQueue->PopFront(); + ReportAsyncDestroyHandleFailed(); + } + + with_lock (HandleOpsQueueLock) { + HandleOpsQueue->PopFront(); + } + RetryDelayedRelease(); + ScheduleProcessHandleOpsQueue(); +} + +void TFileSystem::CompleteAsyncCreateHandle( + TCallContext& callContext, + const NProto::TConfirmCreateHandleResponse& response) +{ + const auto& error = response.GetError(); + RequestStats->RequestCompleted(callContext, error); + + if (HasError(error)) { + STORAGE_ERROR( + "ConfirmCreateHandle request failed: " + << "filesystem " << Config->GetFileSystemId() + << " error: " << FormatError(error)); + if (GetErrorKind(error) == EErrorKind::ErrorRetriable) { + ScheduleProcessHandleOpsQueue(); + return; } - with_lock (DelayedReleaseQueueLock) { - if (!DelayedReleaseQueue.empty()) { - const auto& nextRequest = DelayedReleaseQueue.front(); - if (ProcessAsyncRelease( - nextRequest.CallContext, - nextRequest.Req, - nextRequest.Ino, - nextRequest.Fh, - nextRequest.WriteBackCacheError)) - { - DelayedReleaseQueue.pop(); - } + ReportHandleOpsQueueProcessError( + TStringBuilder() + << "ConfirmCreateHandle failed permanently, filesystem: " + << Config->GetFileSystemId() + << " error: " << FormatError(error)); + } + + with_lock (HandleOpsQueueLock) { + HandleOpsQueue->PopFront(); + } + RetryDelayedRelease(); + ScheduleProcessHandleOpsQueue(); +} + +void TFileSystem::RetryDelayedRelease() +{ + with_lock (DelayedReleaseQueueLock) { + if (!DelayedReleaseQueue.empty()) { + const auto& request = DelayedReleaseQueue.front(); + if (ProcessAsyncRelease( + request.CallContext, + request.Req, + request.Ino, + request.Fh, + request.WriteBackCacheError)) + { + DelayedReleaseQueue.pop(); } } } - ScheduleProcessHandleOpsQueue(); } void TFileSystem::ProcessHandleOpsQueue() { TGuard g{HandleOpsQueueLock}; if (HandleOpsQueue->Empty()) { + // A release can be appended to DelayedReleaseQueue after a queue + // completion has already checked it. Retry here as well to avoid + // leaving that release stranded until another entry completes. + g.Release(); + RetryDelayedRelease(); ScheduleProcessHandleOpsQueue(); return; } @@ -455,8 +492,35 @@ void TFileSystem::ProcessHandleOpsQueue() self->CompleteAsyncDestroyHandle(*callContext, response); } }); + } else if (entry.HasQueuedCreateHandleRequest()) { + const auto& requestInfo = entry.GetQueuedCreateHandleRequest(); + auto request = CreateConfirmRequest( + requestInfo.GetRequest(), + requestInfo.GetNodeId(), + requestInfo.GetHandle(), + requestInfo.GetRequestId()); + + STORAGE_DEBUG( + "Process create handle confirmation request: " + << "filesystem " << Config->GetFileSystemId() << " #" + << requestInfo.GetNodeId() << " @" << requestInfo.GetHandle()); + + auto callContext = MakeIntrusive( + Config->GetFileSystemId(), + CreateRequestId()); + callContext->RequestType = EFileStoreRequest::ConfirmCreateHandle; + RequestStats->RequestStarted(Log, *callContext); + + Session->ConfirmCreateHandle(callContext, std::move(request)) + .Subscribe( + [ptr = weak_from_this(), callContext](const auto& future) + { + const auto& response = future.GetValue(); + if (auto self = ptr.lock()) { + self->CompleteAsyncCreateHandle(*callContext, response); + } + }); } else { - // TODO(#1541): process create handle ReportHandleOpsQueueProcessError( TStringBuilder() << "Unexpected TQueueEntry in queue, filesystem: " << Config->GetFileSystemId()); diff --git a/cloud/filestore/libs/vfs_fuse/fs_impl.h b/cloud/filestore/libs/vfs_fuse/fs_impl.h index 094c2df6d23..55bc03f1aee 100644 --- a/cloud/filestore/libs/vfs_fuse/fs_impl.h +++ b/cloud/filestore/libs/vfs_fuse/fs_impl.h @@ -366,6 +366,29 @@ class TFileSystem final return request; } + static std::shared_ptr + CreateConfirmRequest( + const NProto::TCreateHandleRequest& createRequest, + ui64 nodeId, + ui64 handle, + ui64 requestId); + + void ConfirmCreateHandleForOpen( + TCallContextPtr callContext, + fuse_req_t req, + NProto::TCreateHandleRequest createRequest, + ui64 nodeId, + ui64 handle, + ui64 requestId, + fuse_file_info fi); + + void CleanupFailedCreateHandle( + TCallContextPtr callContext, + fuse_req_t req, + ui64 nodeId, + ui64 handle, + NProto::TError confirmError); + template void SetUserNGroup(T& request, const fuse_ctx* ctx) { @@ -462,6 +485,12 @@ class TFileSystem final const NProto::TNodeAttr& attrs, ui64 version); + void ProcessAsyncCreateHandle( + TCallContextPtr callContext, + fuse_req_t req, + fuse_ino_t ino, + const NProto::TCreateHandleRequest& createRequest, + const NProto::TCreateHandleResponse& response); bool ProcessAsyncRelease( TCallContextPtr callContext, fuse_req_t req, @@ -478,6 +507,10 @@ class TFileSystem final void CompleteAsyncDestroyHandle( TCallContext& callContext, const NProto::TDestroyHandleResponse& response); + void CompleteAsyncCreateHandle( + TCallContext& callContext, + const NProto::TConfirmCreateHandleResponse& response); + void RetryDelayedRelease(); void ClearDirectoryCache(); diff --git a/cloud/filestore/libs/vfs_fuse/fs_impl_data.cpp b/cloud/filestore/libs/vfs_fuse/fs_impl_data.cpp index 6d38acbf679..5f8294c6327 100644 --- a/cloud/filestore/libs/vfs_fuse/fs_impl_data.cpp +++ b/cloud/filestore/libs/vfs_fuse/fs_impl_data.cpp @@ -9,6 +9,7 @@ #include #include +#include #include @@ -45,8 +46,198 @@ void InitNodeInfo( nodeInfo->SetHandle(ToUnderlying(handle)); } +bool IsAsyncCreateHandleEligible( + const TFileSystemConfig& config, + ui32 flags) +{ + // Async create is safe only when close is async too. Otherwise a sync + // DestroyHandle can run before the queued ConfirmCreateHandle. + const bool asyncDestroyHandle = + config.GetAsyncDestroyHandleEnabled() || + config.GetAsyncDestroyReadOnlyHandleEnabled(); + if (!asyncDestroyHandle || !config.GetAsyncCreateHandleEnabled()) { + return false; + } + + constexpr ui32 unsupportedFlags = + ProtoFlag(NProto::TCreateHandleRequest::E_CREATE) | + ProtoFlag(NProto::TCreateHandleRequest::E_WRITE) | + ProtoFlag(NProto::TCreateHandleRequest::E_APPEND) | + ProtoFlag(NProto::TCreateHandleRequest::E_TRUNCATE); + + return (flags & unsupportedFlags) == 0; +} + +fuse_file_info MakeFuseFileInfo( + const NProto::TCreateHandleResponse& response, + const TFileSystemConfig& config) +{ + fuse_file_info fi = {}; + fi.fh = response.GetHandle(); + if (config.GetGuestPageCacheDisabled()) { + fi.direct_io = 1; + } + if (response.GetGuestKeepCache() && config.GetGuestKeepCacheAllowed()) { + fi.keep_cache = 1; + } + + return fi; +} + } // namespace +//////////////////////////////////////////////////////////////////////////////// + +std::shared_ptr +TFileSystem::CreateConfirmRequest( + const NProto::TCreateHandleRequest& createRequest, + ui64 nodeId, + ui64 handle, + ui64 requestId) +{ + auto request = std::make_shared(); + *request->MutableHeaders() = createRequest.GetHeaders(); + request->SetFileSystemId(createRequest.GetFileSystemId()); + request->SetNodeId(nodeId); + request->SetHandle(handle); + request->SetFlags(createRequest.GetFlags()); + request->SetRequestId(requestId); + return request; +} + +void TFileSystem::ConfirmCreateHandleForOpen( + TCallContextPtr callContext, + fuse_req_t req, + NProto::TCreateHandleRequest createRequest, + ui64 nodeId, + ui64 handle, + ui64 requestId, + fuse_file_info fi) +{ + auto confirmRequest = CreateConfirmRequest( + createRequest, + nodeId, + handle, + requestId); + + Session->ConfirmCreateHandle(callContext, std::move(confirmRequest)) + .Subscribe( + [=, ptr = weak_from_this()](const auto& future) mutable + { + auto self = ptr.lock(); + if (!self) { + return; + } + + const auto& confirmResponse = future.GetValue(); + const auto& error = confirmResponse.GetError(); + if (!HasError(error)) { + self->ReplyOpen( + *callContext, + MakeError(S_OK), + req, + &fi); + return; + } + + STORAGE_ERROR( + "ConfirmCreateHandle request failed: " + << "filesystem " << self->Config->GetFileSystemId() + << " error: " << FormatError(error)); + + if (GetErrorKind(error) == EErrorKind::ErrorRetriable) { + self->Scheduler->Schedule( + self->Timer->Now() + + self->Config->GetAsyncHandleOperationPeriod(), + [=, ptr = self->weak_from_this()]() mutable + { + if (auto self = ptr.lock()) { + self->ConfirmCreateHandleForOpen( + callContext, + req, + std::move(createRequest), + nodeId, + handle, + requestId, + fi); + } + }); + return; + } + + self->CleanupFailedCreateHandle( + callContext, + req, + nodeId, + handle, + error); + }); +} + +void TFileSystem::CleanupFailedCreateHandle( + TCallContextPtr callContext, + fuse_req_t req, + ui64 nodeId, + ui64 handle, + NProto::TError confirmError) +{ + auto destroyRequest = StartRequest(nodeId); + destroyRequest->SetHandle(handle); + + Session->DestroyHandle(callContext, std::move(destroyRequest)) + .Subscribe( + [=, ptr = weak_from_this()](const auto& future) mutable + { + auto self = ptr.lock(); + if (!self) { + return; + } + + const auto& destroyResponse = future.GetValue(); + const auto& destroyError = destroyResponse.GetError(); + if (HasError(destroyError)) { + STORAGE_ERROR( + "DestroyHandle after failed ConfirmCreateHandle " + "failed: filesystem " + << self->Config->GetFileSystemId() + << " error: " << FormatError(destroyError)); + + if (GetErrorKind(destroyError) == + EErrorKind::ErrorRetriable) + { + self->Scheduler->Schedule( + self->Timer->Now() + + self->Config->GetAsyncHandleOperationPeriod(), + [=, ptr = self->weak_from_this()]() mutable + { + if (auto self = ptr.lock()) { + self->CleanupFailedCreateHandle( + callContext, + req, + nodeId, + handle, + confirmError); + } + }); + return; + } + + ReportHandleOpsQueueProcessError( + TStringBuilder() + << "DestroyHandle after failed ConfirmCreateHandle " + << "failed permanently, filesystem: " + << self->Config->GetFileSystemId() + << " error: " << FormatError(destroyError)); + } + + self->ReplyError( + *callContext, + confirmError, + req, + ErrnoFromError(confirmError.GetCode())); + }); +} + //////////////////////////////////////////////////////////////////////////////// // zero-copy iovec helpers @@ -209,28 +400,105 @@ void TFileSystem::Open( : 0; request->SetFlags(flags | overrideRead); + const bool asyncCreateHandle = + HandleOpsQueue && + IsAsyncCreateHandleEligible(*Config, request->GetFlags()); + if (asyncCreateHandle) { + request->SetAllowAsyncCreateHandle(true); + } + auto requestForQueue = request; + Session->CreateHandle(callContext, std::move(request)) .Subscribe([=, ptr = weak_from_this()] (const auto& future) { const auto& response = future.GetValue(); if (auto self = ptr.lock(); CheckResponse(self, *callContext, req, response)) { const auto& response = future.GetValue(); - fuse_file_info fi = {}; - fi.fh = response.GetHandle(); - if (self->Config->GetGuestPageCacheDisabled()) { - fi.direct_io = 1; - } - if (response.GetGuestKeepCache() && - self->Config->GetGuestKeepCacheAllowed()) - { - fi.keep_cache = 1; + if (response.GetHandleCreatedAsync()) { + STORAGE_VERIFY_C( + asyncCreateHandle, + TWellKnownEntityTypes::FILESYSTEM, + self->Config->GetFileSystemId(), + TStringBuilder() + << "CreateHandle returned an unpersisted handle " + << "without client-side async create support #" + << ino << " @" << response.GetHandle()); + self->ProcessAsyncCreateHandle( + callContext, + req, + ino, + *requestForQueue, + response); + return; } + auto fi = MakeFuseFileInfo(response, *self->Config); self->ReplyOpen(*callContext, response.GetError(), req, &fi); } }); } +void TFileSystem::ProcessAsyncCreateHandle( + TCallContextPtr callContext, + fuse_req_t req, + fuse_ino_t ino, + const NProto::TCreateHandleRequest& createRequest, + const NProto::TCreateHandleResponse& response) +{ + const auto requestId = createRequest.GetHeaders().GetRequestId() + ? createRequest.GetHeaders().GetRequestId() + : callContext->RequestId; + + THandleOpsQueue::EResult result; + with_lock (HandleOpsQueueLock) { + result = HandleOpsQueue->AddCreateRequest( + createRequest, + ino, + response.GetHandle(), + requestId); + } + + if (result == THandleOpsQueue::EResult::Ok) { + STORAGE_DEBUG( + "Create handle request added to queue #" + << ino << " @" << response.GetHandle()); + auto fi = MakeFuseFileInfo(response, *Config); + ReplyOpen(*callContext, MakeError(S_OK), req, &fi); + return; + } + + if (result == THandleOpsQueue::EResult::SerializationError) { + TStringBuilder msg; + msg << "Unable to add CreateHandleRequest to " + << "HandleOpsQueue #" << ino << " @" + << response.GetHandle() + << ". Serialization failed"; + ReportHandleOpsQueueProcessError(msg); + + auto destroyRequest = StartRequest(ino); + destroyRequest->SetHandle(response.GetHandle()); + Session->DestroyHandle(callContext, std::move(destroyRequest)); + ReplyError(*callContext, MakeError(E_FAIL, msg), req, EIO); + return; + } + + // result == THandleOpsQueue::EResult::QueueOverflow + STORAGE_DEBUG( + "HandleOpsQueue overflow, synchronously " + "confirming create handle #" + << ino << " @" << response.GetHandle()); + + const auto fi = MakeFuseFileInfo(response, *Config); + ConfirmCreateHandleForOpen( + std::move(callContext), + req, + createRequest, + ino, + response.GetHandle(), + requestId, + fi); +} + bool TFileSystem::ProcessAsyncRelease( TCallContextPtr callContext, fuse_req_t req, diff --git a/cloud/filestore/libs/vfs_fuse/fs_ut.cpp b/cloud/filestore/libs/vfs_fuse/fs_ut.cpp index 33437aec391..17851b3f4b8 100644 --- a/cloud/filestore/libs/vfs_fuse/fs_ut.cpp +++ b/cloud/filestore/libs/vfs_fuse/fs_ut.cpp @@ -20,6 +20,7 @@ #include #include #include +#include #include #include #include @@ -260,7 +261,8 @@ struct TBootstrap proto.SetSocketPath(SocketPath); proto.SetFileSystemId(FileSystemId); if (featuresConfig.GetAsyncDestroyHandleEnabled() || - featuresConfig.GetAsyncDestroyReadOnlyHandleEnabled()) + featuresConfig.GetAsyncDestroyReadOnlyHandleEnabled() || + featuresConfig.GetAsyncCreateHandleEnabled()) { proto.SetHandleOpsQueuePath(TempDir.Path() / "HandleOpsQueue"); proto.SetHandleOpsQueueSize(handleOpsQueueSize); @@ -3195,6 +3197,799 @@ Y_UNIT_TEST_SUITE(TFileSystemTest) AtomicGet(counters->GetCounter("InProgress")->GetAtomic())); } + Y_UNIT_TEST(ShouldNotSetAsyncCreateHandleIfFeatureDisabled) + { + // The vhost-side feature gate is off, so the client must not advertise + // async create support even for an otherwise eligible read-only open. + auto scheduler = std::make_shared(); + TBootstrap bootstrap(CreateWallClockTimer(), scheduler); + + const ui64 nodeId = 10; + const ui64 handle = 2; + std::atomic_uint confirmCalled = 0; + + bootstrap.Service->SetHandlerCreateHandle( + [&](auto callContext, auto request) + { + UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId); + UNIT_ASSERT_VALUES_EQUAL(nodeId, request->GetNodeId()); + UNIT_ASSERT(!request->GetAllowAsyncCreateHandle()); + + NProto::TCreateHandleResponse response; + response.SetHandle(handle); + response.MutableNodeAttr()->SetId(nodeId); + response.MutableNodeAttr()->SetType(NProto::E_REGULAR_NODE); + return MakeFuture(response); + }); + + bootstrap.Service->SetHandlerConfirmCreateHandle( + [&](auto, auto) + { + ++confirmCalled; + return MakeFuture(NProto::TConfirmCreateHandleResponse{}); + }); + + bootstrap.Start(); + Y_DEFER { + bootstrap.Stop(); + }; + + auto future = bootstrap.Fuse->SendRequest(nodeId); + UNIT_ASSERT_VALUES_EQUAL(handle, future.GetValue(WaitTimeout)); + + scheduler->RunAllScheduledTasks(); + UNIT_ASSERT_VALUES_EQUAL(0U, confirmCalled.load()); + } + + Y_UNIT_TEST(ShouldNotSetAsyncCreateHandleIfAsyncDestroyDisabled) + { + NProto::TFileStoreFeatures features; + features.SetAsyncCreateHandleEnabled(true); + auto scheduler = std::make_shared(); + TBootstrap bootstrap(CreateWallClockTimer(), scheduler, features); + + const ui64 nodeId = 10; + const ui64 handle = 2; + std::atomic_uint confirmCalled = 0; + + bootstrap.Service->SetHandlerCreateHandle( + [&](auto callContext, auto request) + { + UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId); + UNIT_ASSERT_VALUES_EQUAL(nodeId, request->GetNodeId()); + UNIT_ASSERT(!request->GetAllowAsyncCreateHandle()); + + NProto::TCreateHandleResponse response; + response.SetHandle(handle); + response.MutableNodeAttr()->SetId(nodeId); + response.MutableNodeAttr()->SetType(NProto::E_REGULAR_NODE); + return MakeFuture(response); + }); + + bootstrap.Service->SetHandlerConfirmCreateHandle( + [&](auto, auto) + { + ++confirmCalled; + return MakeFuture(NProto::TConfirmCreateHandleResponse{}); + }); + + bootstrap.Start(); + Y_DEFER { + bootstrap.Stop(); + }; + + auto future = bootstrap.Fuse->SendRequest(nodeId); + UNIT_ASSERT_VALUES_EQUAL(handle, future.GetValue(WaitTimeout)); + + scheduler->RunAllScheduledTasks(); + UNIT_ASSERT_VALUES_EQUAL(0U, confirmCalled.load()); + } + + void CheckIneligibleRequestDoesNotUseAsyncCreate( + bool create, + int systemFlags, + ui32 expectedProtoFlag) + { + // Async create is allowed only for plain read-only opens. Create, + // write, and truncate requests must stay on the synchronous path. + NProto::TFileStoreFeatures features; + features.SetAsyncCreateHandleEnabled(true); + features.SetAsyncDestroyReadOnlyHandleEnabled(true); + auto scheduler = std::make_shared(); + TBootstrap bootstrap(CreateWallClockTimer(), scheduler, features); + + const ui64 nodeId = 10; + const ui64 handle = 2; + std::atomic_uint confirmCalled = 0; + + bootstrap.Service->SetHandlerCreateHandle( + [&](auto callContext, auto request) + { + UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId); + UNIT_ASSERT_VALUES_EQUAL(nodeId, request->GetNodeId()); + UNIT_ASSERT(request->GetFlags() & expectedProtoFlag); + UNIT_ASSERT(!request->GetAllowAsyncCreateHandle()); + + NProto::TCreateHandleResponse response; + response.SetHandle(handle); + response.MutableNodeAttr()->SetId(nodeId); + response.MutableNodeAttr()->SetType(NProto::E_REGULAR_NODE); + return MakeFuture(response); + }); + + bootstrap.Service->SetHandlerConfirmCreateHandle( + [&](auto, auto) + { + ++confirmCalled; + return MakeFuture(NProto::TConfirmCreateHandleResponse{}); + }); + + bootstrap.Start(); + Y_DEFER { + bootstrap.Stop(); + }; + + TFuture future; + if (create) { + auto request = + std::make_shared("/file1", nodeId); + request->In->Body.flags |= systemFlags; + future = bootstrap.Fuse->SendRequest(request); + } else { + auto request = std::make_shared(nodeId); + request->In->Body.flags |= systemFlags; + future = bootstrap.Fuse->SendRequest(request); + } + + UNIT_ASSERT_VALUES_EQUAL(handle, future.GetValue(WaitTimeout)); + + scheduler->RunAllScheduledTasks(); + UNIT_ASSERT_VALUES_EQUAL(0U, confirmCalled.load()); + } + + Y_UNIT_TEST(ShouldNotSetAsyncCreateHandleForIneligibleRequests) + { + CheckIneligibleRequestDoesNotUseAsyncCreate( + false, // open + O_WRONLY, + ProtoFlag(NProto::TCreateHandleRequest::E_WRITE)); + CheckIneligibleRequestDoesNotUseAsyncCreate( + false, // open + O_TRUNC, + ProtoFlag(NProto::TCreateHandleRequest::E_TRUNCATE)); + CheckIneligibleRequestDoesNotUseAsyncCreate( + true, // create + O_CREAT, + ProtoFlag(NProto::TCreateHandleRequest::E_CREATE)); + } + + Y_UNIT_TEST(ShouldNotQueueAsyncCreateHandleForPersistedResponse) + { + // The request advertises async support, but the server can still return + // a persisted handle. In that case there is nothing to confirm later. + NProto::TFileStoreFeatures features; + features.SetAsyncCreateHandleEnabled(true); + features.SetAsyncDestroyReadOnlyHandleEnabled(true); + auto scheduler = std::make_shared(); + TBootstrap bootstrap(CreateWallClockTimer(), scheduler, features); + + const ui64 nodeId = 10; + const ui64 handle = 2; + std::atomic_uint confirmCalled = 0; + + bootstrap.Service->SetHandlerCreateHandle( + [&](auto callContext, auto request) + { + UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId); + UNIT_ASSERT_VALUES_EQUAL(nodeId, request->GetNodeId()); + UNIT_ASSERT(request->GetAllowAsyncCreateHandle()); + + NProto::TCreateHandleResponse response; + response.SetHandle(handle); + response.MutableNodeAttr()->SetId(nodeId); + response.MutableNodeAttr()->SetType(NProto::E_REGULAR_NODE); + return MakeFuture(response); + }); + + bootstrap.Service->SetHandlerConfirmCreateHandle( + [&](auto, auto) + { + ++confirmCalled; + return MakeFuture(NProto::TConfirmCreateHandleResponse{}); + }); + + bootstrap.Start(); + Y_DEFER { + bootstrap.Stop(); + }; + + auto future = bootstrap.Fuse->SendRequest(nodeId); + UNIT_ASSERT_VALUES_EQUAL(handle, future.GetValue(WaitTimeout)); + + scheduler->RunAllScheduledTasks(); + UNIT_ASSERT_VALUES_EQUAL(0U, confirmCalled.load()); + } + + Y_UNIT_TEST(ShouldQueueAsyncCreateHandleBeforeReplyingOpen) + { + NProto::TFileStoreFeatures features; + features.SetAsyncCreateHandleEnabled(true); + features.SetAsyncDestroyReadOnlyHandleEnabled(true); + auto scheduler = std::make_shared(); + TBootstrap bootstrap(CreateWallClockTimer(), scheduler, features); + + const ui64 nodeId = 10; + const ui64 handle = 2; + std::atomic_bool openFinished = false; + std::atomic_uint confirmCalled = 0; + + bootstrap.Service->SetHandlerCreateHandle( + [&](auto callContext, auto request) + { + UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId); + UNIT_ASSERT_VALUES_EQUAL(nodeId, request->GetNodeId()); + UNIT_ASSERT(request->GetAllowAsyncCreateHandle()); + + NProto::TCreateHandleResponse response; + response.SetHandle(handle); + response.SetHandleCreatedAsync(true); + response.MutableNodeAttr()->SetId(nodeId); + response.MutableNodeAttr()->SetType(NProto::E_REGULAR_NODE); + return MakeFuture(response); + }); + + bootstrap.Service->SetHandlerConfirmCreateHandle( + [&](auto callContext, auto request) + { + UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId); + UNIT_ASSERT(openFinished); + UNIT_ASSERT_VALUES_EQUAL(nodeId, request->GetNodeId()); + UNIT_ASSERT_VALUES_EQUAL(handle, request->GetHandle()); + UNIT_ASSERT_VALUES_EQUAL( + SessionId, + request->GetHeaders().GetSessionId()); + ++confirmCalled; + return MakeFuture(NProto::TConfirmCreateHandleResponse{}); + }); + + bootstrap.Start(); + Y_DEFER { + bootstrap.Stop(); + }; + + auto future = bootstrap.Fuse->SendRequest(nodeId); + UNIT_ASSERT_VALUES_EQUAL(handle, future.GetValue(WaitTimeout)); + openFinished = true; + // ConfirmCreateHandle is driven by scheduled queue processing, not by + // the foreground open path. + UNIT_ASSERT_VALUES_EQUAL(0U, confirmCalled.load()); + + scheduler->RunAllScheduledTasks(); + UNIT_ASSERT_VALUES_EQUAL(1U, confirmCalled.load()); + } + + Y_UNIT_TEST(ShouldRetryAsyncCreateHandleConfirmationOnRetriableError) + { + NProto::TFileStoreFeatures features; + features.SetAsyncCreateHandleEnabled(true); + features.SetAsyncDestroyReadOnlyHandleEnabled(true); + auto scheduler = std::make_shared(); + TBootstrap bootstrap(CreateWallClockTimer(), scheduler, features); + + const ui64 nodeId = 10; + const ui64 handle = 2; + std::atomic_bool openFinished = false; + std::atomic_uint confirmCalled = 0; + + bootstrap.Service->SetHandlerCreateHandle( + [&](auto callContext, auto request) + { + UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId); + UNIT_ASSERT_VALUES_EQUAL(nodeId, request->GetNodeId()); + UNIT_ASSERT(request->GetAllowAsyncCreateHandle()); + + NProto::TCreateHandleResponse response; + response.SetHandle(handle); + response.SetHandleCreatedAsync(true); + response.MutableNodeAttr()->SetId(nodeId); + response.MutableNodeAttr()->SetType(NProto::E_REGULAR_NODE); + return MakeFuture(response); + }); + + bootstrap.Service->SetHandlerConfirmCreateHandle( + [&](auto callContext, auto request) + { + UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId); + UNIT_ASSERT(openFinished); + UNIT_ASSERT_VALUES_EQUAL(nodeId, request->GetNodeId()); + UNIT_ASSERT_VALUES_EQUAL(handle, request->GetHandle()); + UNIT_ASSERT_VALUES_EQUAL( + SessionId, + request->GetHeaders().GetSessionId()); + + if (++confirmCalled < 3) { + NProto::TConfirmCreateHandleResponse response = + TErrorResponse(E_REJECTED, "retriable"); + return MakeFuture(response); + } + + return MakeFuture(NProto::TConfirmCreateHandleResponse{}); + }); + + bootstrap.Start(); + Y_DEFER { + bootstrap.Stop(); + }; + + auto future = bootstrap.Fuse->SendRequest(nodeId); + UNIT_ASSERT_VALUES_EQUAL(handle, future.GetValue(WaitTimeout)); + openFinished = true; + + scheduler->RunAllScheduledTasks(); + UNIT_ASSERT_VALUES_EQUAL(1U, confirmCalled.load()); + + scheduler->RunAllScheduledTasks(); + UNIT_ASSERT_VALUES_EQUAL(2U, confirmCalled.load()); + + scheduler->RunAllScheduledTasks(); + UNIT_ASSERT_VALUES_EQUAL(3U, confirmCalled.load()); + + // The successful third confirmation pops the queue entry; later queue + // processing must not retry it. + scheduler->RunAllScheduledTasks(); + UNIT_ASSERT_VALUES_EQUAL(3U, confirmCalled.load()); + } + + Y_UNIT_TEST(ShouldNotRetryAsyncCreateHandleConfirmationOnFinalError) + { + NProto::TFileStoreFeatures features; + features.SetAsyncCreateHandleEnabled(true); + features.SetAsyncDestroyReadOnlyHandleEnabled(true); + auto scheduler = std::make_shared(); + TBootstrap bootstrap(CreateWallClockTimer(), scheduler, features); + + const ui64 nodeId = 10; + const ui64 handle = 2; + std::atomic_bool openFinished = false; + std::atomic_uint confirmCalled = 0; + + bootstrap.Service->SetHandlerCreateHandle( + [&](auto callContext, auto request) + { + UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId); + UNIT_ASSERT_VALUES_EQUAL(nodeId, request->GetNodeId()); + UNIT_ASSERT(request->GetAllowAsyncCreateHandle()); + + NProto::TCreateHandleResponse response; + response.SetHandle(handle); + response.SetHandleCreatedAsync(true); + response.MutableNodeAttr()->SetId(nodeId); + response.MutableNodeAttr()->SetType(NProto::E_REGULAR_NODE); + return MakeFuture(response); + }); + + bootstrap.Service->SetHandlerConfirmCreateHandle( + [&](auto callContext, auto request) + { + UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId); + UNIT_ASSERT(openFinished); + UNIT_ASSERT_VALUES_EQUAL(nodeId, request->GetNodeId()); + UNIT_ASSERT_VALUES_EQUAL(handle, request->GetHandle()); + UNIT_ASSERT_VALUES_EQUAL( + SessionId, + request->GetHeaders().GetSessionId()); + + ++confirmCalled; + NProto::TConfirmCreateHandleResponse response = + TErrorResponse(E_FS_NOENT, "final"); + return MakeFuture(response); + }); + + bootstrap.Start(); + Y_DEFER { + bootstrap.Stop(); + }; + + auto future = bootstrap.Fuse->SendRequest(nodeId); + UNIT_ASSERT_VALUES_EQUAL(handle, future.GetValue(WaitTimeout)); + openFinished = true; + + scheduler->RunAllScheduledTasks(); + UNIT_ASSERT_VALUES_EQUAL(1U, confirmCalled.load()); + + auto errorCounter = + bootstrap.Counters->GetSubgroup("component", "fs_ut") + ->GetCounter( + "AppCriticalEvents/HandleOpsQueueProcessError", + true); + UNIT_ASSERT_VALUES_EQUAL(1, static_cast(*errorCounter)); + + // A final confirmation error pops the queue entry, so later queue + // processing must not retry it. + scheduler->RunAllScheduledTasks(); + UNIT_ASSERT_VALUES_EQUAL(1U, confirmCalled.load()); + } + + Y_UNIT_TEST(ShouldConfirmAsyncCreateHandleSynchronouslyIfQueueOverflows) + { + NProto::TFileStoreFeatures features; + features.SetAsyncCreateHandleEnabled(true); + features.SetAsyncDestroyReadOnlyHandleEnabled(true); + auto scheduler = std::make_shared(); + // Keep the ring buffer smaller than a serialized queued create entry so + // AddCreateRequest returns QueueOverflow. + TBootstrap bootstrap(CreateWallClockTimer(), scheduler, features, 20); + + const ui64 nodeId = 10; + const ui64 handle = 2; + std::atomic_uint confirmCalled = 0; + auto confirmPromise = NewPromise(); + + bootstrap.Service->SetHandlerCreateHandle( + [&](auto callContext, auto request) + { + UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId); + UNIT_ASSERT_VALUES_EQUAL(nodeId, request->GetNodeId()); + UNIT_ASSERT(request->GetAllowAsyncCreateHandle()); + + NProto::TCreateHandleResponse response; + response.SetHandle(handle); + response.SetHandleCreatedAsync(true); + response.MutableNodeAttr()->SetId(nodeId); + response.MutableNodeAttr()->SetType(NProto::E_REGULAR_NODE); + return MakeFuture(response); + }); + + bootstrap.Service->SetHandlerConfirmCreateHandle( + [&, confirmPromise](auto callContext, auto request) + { + UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId); + UNIT_ASSERT_VALUES_EQUAL(nodeId, request->GetNodeId()); + UNIT_ASSERT_VALUES_EQUAL(handle, request->GetHandle()); + UNIT_ASSERT_VALUES_EQUAL( + SessionId, + request->GetHeaders().GetSessionId()); + ++confirmCalled; + return confirmPromise; + }); + + bootstrap.Start(); + Y_DEFER { + bootstrap.Stop(); + }; + + auto future = bootstrap.Fuse->SendRequest(nodeId); + // Queue overflow forces synchronous confirmation before ReplyOpen, so + // the FUSE request must stay pending while confirmPromise is unresolved. + UNIT_ASSERT_EXCEPTION( + future.GetValue(ExceptionWaitTimeout), + yexception); + UNIT_ASSERT_VALUES_EQUAL(1U, confirmCalled.load()); + + confirmPromise.SetValue(NProto::TConfirmCreateHandleResponse{}); + UNIT_ASSERT_VALUES_EQUAL(handle, future.GetValue(WaitTimeout)); + + // The create was not queued, so background queue processing must not + // issue another ConfirmCreateHandle. + scheduler->RunAllScheduledTasks(); + UNIT_ASSERT_VALUES_EQUAL(1U, confirmCalled.load()); + } + + Y_UNIT_TEST(ShouldRetryOverflowAsyncCreateConfirmationOnRetriableError) + { + NProto::TFileStoreFeatures features; + features.SetAsyncCreateHandleEnabled(true); + features.SetAsyncDestroyReadOnlyHandleEnabled(true); + auto scheduler = std::make_shared(); + // Keep the ring buffer smaller than a serialized queued create entry so + // AddCreateRequest returns QueueOverflow. + TBootstrap bootstrap(CreateWallClockTimer(), scheduler, features, 20); + + const ui64 nodeId = 10; + const ui64 handle = 2; + std::atomic_uint confirmCalled = 0; + + bootstrap.Service->SetHandlerCreateHandle( + [&](auto callContext, auto request) + { + UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId); + UNIT_ASSERT_VALUES_EQUAL(nodeId, request->GetNodeId()); + UNIT_ASSERT(request->GetAllowAsyncCreateHandle()); + + NProto::TCreateHandleResponse response; + response.SetHandle(handle); + response.SetHandleCreatedAsync(true); + response.MutableNodeAttr()->SetId(nodeId); + response.MutableNodeAttr()->SetType(NProto::E_REGULAR_NODE); + return MakeFuture(response); + }); + + bootstrap.Service->SetHandlerConfirmCreateHandle( + [&](auto callContext, auto request) + { + UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId); + UNIT_ASSERT_VALUES_EQUAL(nodeId, request->GetNodeId()); + UNIT_ASSERT_VALUES_EQUAL(handle, request->GetHandle()); + UNIT_ASSERT_VALUES_EQUAL( + SessionId, + request->GetHeaders().GetSessionId()); + + if (++confirmCalled < 3) { + NProto::TConfirmCreateHandleResponse response = + TErrorResponse(E_REJECTED, "retriable"); + return MakeFuture(response); + } + + return MakeFuture(NProto::TConfirmCreateHandleResponse{}); + }); + + bootstrap.Start(); + Y_DEFER { + bootstrap.Stop(); + }; + + auto future = bootstrap.Fuse->SendRequest(nodeId); + UNIT_ASSERT_EXCEPTION( + future.GetValue(ExceptionWaitTimeout), + yexception); + UNIT_ASSERT_VALUES_EQUAL(1U, confirmCalled.load()); + + scheduler->RunAllScheduledTasks(); + UNIT_ASSERT_EXCEPTION( + future.GetValue(ExceptionWaitTimeout), + yexception); + UNIT_ASSERT_VALUES_EQUAL(2U, confirmCalled.load()); + + scheduler->RunAllScheduledTasks(); + UNIT_ASSERT_VALUES_EQUAL(handle, future.GetValue(WaitTimeout)); + UNIT_ASSERT_VALUES_EQUAL(3U, confirmCalled.load()); + } + + Y_UNIT_TEST(ShouldFailOpenIfOverflowAsyncCreateConfirmationFails) + { + NProto::TFileStoreFeatures features; + features.SetAsyncCreateHandleEnabled(true); + features.SetAsyncDestroyReadOnlyHandleEnabled(true); + auto scheduler = std::make_shared(); + // Keep the ring buffer smaller than a serialized queued create entry so + // AddCreateRequest returns QueueOverflow. + TBootstrap bootstrap(CreateWallClockTimer(), scheduler, features, 20); + + const ui64 nodeId = 10; + const ui64 handle = 2; + std::atomic_uint confirmCalled = 0; + std::atomic_uint destroyCalled = 0; + auto confirmPromise = NewPromise(); + auto destroyPromise = NewPromise(); + + bootstrap.Service->SetHandlerCreateHandle( + [&](auto callContext, auto request) + { + UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId); + UNIT_ASSERT_VALUES_EQUAL(nodeId, request->GetNodeId()); + UNIT_ASSERT(request->GetAllowAsyncCreateHandle()); + + NProto::TCreateHandleResponse response; + response.SetHandle(handle); + response.SetHandleCreatedAsync(true); + response.MutableNodeAttr()->SetId(nodeId); + response.MutableNodeAttr()->SetType(NProto::E_REGULAR_NODE); + return MakeFuture(response); + }); + + bootstrap.Service->SetHandlerConfirmCreateHandle( + [&, confirmPromise](auto callContext, auto request) + { + UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId); + UNIT_ASSERT_VALUES_EQUAL(nodeId, request->GetNodeId()); + UNIT_ASSERT_VALUES_EQUAL(handle, request->GetHandle()); + UNIT_ASSERT_VALUES_EQUAL( + SessionId, + request->GetHeaders().GetSessionId()); + ++confirmCalled; + return confirmPromise; + }); + + bootstrap.Service->SetHandlerDestroyHandle( + [&, destroyPromise](auto callContext, auto request) + { + UNIT_ASSERT_VALUES_EQUAL(FileSystemId, callContext->FileSystemId); + UNIT_ASSERT_VALUES_EQUAL(nodeId, request->GetNodeId()); + UNIT_ASSERT_VALUES_EQUAL(handle, request->GetHandle()); + ++destroyCalled; + return destroyPromise; + }); + + bootstrap.Start(); + Y_DEFER { + bootstrap.Stop(); + }; + + auto future = bootstrap.Fuse->SendRequest(nodeId); + UNIT_ASSERT_EXCEPTION( + future.GetValue(ExceptionWaitTimeout), + yexception); + UNIT_ASSERT_VALUES_EQUAL(1U, confirmCalled.load()); + + NProto::TConfirmCreateHandleResponse response = + TErrorResponse(E_FS_NOENT, "final"); + confirmPromise.SetValue(std::move(response)); + UNIT_ASSERT_EXCEPTION( + future.GetValue(ExceptionWaitTimeout), + yexception); + UNIT_ASSERT_VALUES_EQUAL(1U, destroyCalled.load()); + + destroyPromise.SetValue(NProto::TDestroyHandleResponse{}); + UNIT_ASSERT_EXCEPTION_CONTAINS( + future.GetValue(WaitTimeout), + yexception, + ToString(-ENOENT)); + + // The create was never queued, so background queue processing must not + // retry the failed synchronous confirmation. + scheduler->RunAllScheduledTasks(); + UNIT_ASSERT_VALUES_EQUAL(1U, confirmCalled.load()); + } + + Y_UNIT_TEST(ShouldRestoreAndDrainAsyncCreateHandleQueueAfterSessionRestart) + { + const TString sessionId = CreateGuidAsString(); + + NProto::TFileStoreFeatures features; + features.SetAsyncCreateHandleEnabled(true); + features.SetAsyncDestroyReadOnlyHandleEnabled(true); + + const ui64 nodeId = 10; + const ui64 handle = 2; + std::atomic requestId = 0; + std::atomic_uint confirmCalled = 0; + + auto createBootstrap = [&]() + { + TBootstrap bootstrap( + CreateWallClockTimer(), + std::make_shared(TInstant::Zero()), + features); + + bootstrap.Service->CreateSessionHandler = + [features, sessionId](auto, auto) + { + NProto::TCreateSessionResponse result; + result.MutableSession()->SetSessionId(sessionId); + result.MutableFileStore()->SetBlockSize(4096); + result.MutableFileStore()->MutableFeatures()->CopyFrom( + features); + result.MutableFileStore()->SetFileSystemId(FileSystemId); + return MakeFuture(result); + }; + + return bootstrap; + }; + + { + auto bootstrap = createBootstrap(); + + bootstrap.Service->SetHandlerCreateHandle( + [&](auto callContext, auto request) + { + UNIT_ASSERT_VALUES_EQUAL( + FileSystemId, + callContext->FileSystemId); + UNIT_ASSERT_VALUES_EQUAL(nodeId, request->GetNodeId()); + UNIT_ASSERT(request->GetAllowAsyncCreateHandle()); + + NProto::TCreateHandleResponse response; + response.SetHandle(handle); + response.SetHandleCreatedAsync(true); + response.MutableNodeAttr()->SetId(nodeId); + response.MutableNodeAttr()->SetType( + NProto::E_REGULAR_NODE); + return MakeFuture(response); + }); + + bootstrap.Service->SetHandlerConfirmCreateHandle( + [&](auto callContext, auto request) + { + UNIT_ASSERT_VALUES_EQUAL( + FileSystemId, + callContext->FileSystemId); + UNIT_ASSERT_VALUES_EQUAL(nodeId, request->GetNodeId()); + UNIT_ASSERT_VALUES_EQUAL(handle, request->GetHandle()); + UNIT_ASSERT_VALUES_EQUAL( + sessionId, + request->GetHeaders().GetSessionId()); + UNIT_ASSERT(request->GetRequestId()); + UNIT_ASSERT_VALUES_EQUAL( + ProtoFlag(NProto::TCreateHandleRequest::E_READ), + request->GetFlags()); + requestId = request->GetRequestId(); + ++confirmCalled; + + NProto::TConfirmCreateHandleResponse response = + TErrorResponse(E_REJECTED, "retriable"); + return MakeFuture(response); + }); + + bootstrap.Start(); + Y_DEFER { + bootstrap.Stop(); + }; + + auto future = + bootstrap.Fuse->SendRequest(nodeId); + UNIT_ASSERT_VALUES_EQUAL(handle, future.GetValue(WaitTimeout)); + + auto* scheduler = + dynamic_cast(bootstrap.Scheduler.get()); + UNIT_ASSERT(scheduler); + scheduler->RunAllScheduledTasks(); + + // The retriable failure leaves the durable queue entry in place for + // the restarted loop to reload and confirm. + UNIT_ASSERT_VALUES_EQUAL(1U, confirmCalled.load()); + UNIT_ASSERT(requestId.load()); + + auto suspend = bootstrap.Loop->SuspendAsync(); + UNIT_ASSERT(WaitForCondition( + WaitTimeout, + [&] + { + scheduler->RunAllScheduledTasksUntilNow(); + return suspend.HasValue() || suspend.HasException(); + })); + UNIT_ASSERT_NO_EXCEPTION(suspend.GetValue(WaitTimeout)); + + // A clean Stop() destroys the session and removes the queue. Drop + // the loop after suspend to model a vhost restart with the session + // and queue directory still present. + bootstrap.Loop = nullptr; + } + + { + auto bootstrap = createBootstrap(); + + bootstrap.Service->SetHandlerConfirmCreateHandle( + [&](auto callContext, auto request) + { + UNIT_ASSERT_VALUES_EQUAL( + FileSystemId, + callContext->FileSystemId); + UNIT_ASSERT_VALUES_EQUAL(nodeId, request->GetNodeId()); + UNIT_ASSERT_VALUES_EQUAL(handle, request->GetHandle()); + UNIT_ASSERT_VALUES_EQUAL( + sessionId, + request->GetHeaders().GetSessionId()); + UNIT_ASSERT_VALUES_EQUAL( + requestId.load(), + request->GetRequestId()); + UNIT_ASSERT_VALUES_EQUAL( + ProtoFlag(NProto::TCreateHandleRequest::E_READ), + request->GetFlags()); + ++confirmCalled; + return MakeFuture(NProto::TConfirmCreateHandleResponse{}); + }); + + bootstrap.Start(); + Y_DEFER { + bootstrap.Stop(); + }; + + UNIT_ASSERT_VALUES_EQUAL(1U, confirmCalled.load()); + + auto* scheduler = + dynamic_cast(bootstrap.Scheduler.get()); + UNIT_ASSERT(scheduler); + scheduler->RunAllScheduledTasks(); + + UNIT_ASSERT_VALUES_EQUAL(2U, confirmCalled.load()); + + // The restarted loop drained and popped the durable entry, so + // another scheduling pass must not confirm the same handle again. + scheduler->RunAllScheduledTasks(); + UNIT_ASSERT_VALUES_EQUAL(2U, confirmCalled.load()); + } + } + Y_UNIT_TEST(ShouldRetryDestroyIfNotSuccessDuringAsyncProcessing) { NProto::TFileStoreFeatures features; @@ -3405,12 +4200,13 @@ Y_UNIT_TEST_SUITE(TFileSystemTest) 1, AtomicGet(counters->GetCounter("InProgress")->GetAtomic())); - // Process first request. + // A final error still pops the first entry, so it must also retry the + // postponed release. scheduler->RunAllScheduledTasks(); - responsePromise.SetValue(NProto::TDestroyHandleResponse{}); + responsePromise.SetValue(TErrorResponse(E_FS_NOENT, "final")); UNIT_ASSERT_VALUES_EQUAL(1u, handlerCalled.load()); - // After the first request is processed, the second request should be + // After the first request is removed, the second request should be // completed and added to the HandleOpsQueue. UNIT_ASSERT_NO_EXCEPTION(future.GetValue(WaitTimeout)); UNIT_ASSERT_VALUES_EQUAL( @@ -3422,6 +4218,136 @@ Y_UNIT_TEST_SUITE(TFileSystemTest) UNIT_ASSERT_VALUES_EQUAL(2u, handlerCalled.load()); } + Y_UNIT_TEST(ShouldRetryPostponedReleaseAfterAsyncCreateHandleConfirmation) + { + NProto::TFileStoreFeatures features; + features.SetAsyncCreateHandleEnabled(true); + features.SetAsyncDestroyHandleEnabled(true); + + const ui64 nodeId = 10; + const ui64 handle = 2; + auto asyncResponse = [=] + { + NProto::TCreateHandleResponse response; + response.SetHandle(handle); + response.SetHandleCreatedAsync(true); + response.MutableNodeAttr()->SetId(nodeId); + response.MutableNodeAttr()->SetType(NProto::E_REGULAR_NODE); + return response; + }; + NProto::TCreateHandleRequest createRequest; + + { + auto scheduler = std::make_shared(); + TBootstrap bootstrap(CreateWallClockTimer(), scheduler, features); + bootstrap.Service->SetHandlerCreateHandle( + [&](auto, auto request) + { + createRequest = *request; + return MakeFuture(asyncResponse()); + }); + bootstrap.Service->SetHandlerConfirmCreateHandle( + [](auto, auto) + { + return MakeFuture(NProto::TConfirmCreateHandleResponse{}); + }); + + bootstrap.Start(); + Y_DEFER { + bootstrap.Stop(); + }; + UNIT_ASSERT_VALUES_EQUAL( + handle, + bootstrap.Fuse->SendRequest(nodeId) + .GetValue(WaitTimeout)); + scheduler->RunAllScheduledTasks(); + } + + NProto::TQueueEntry createEntry; + auto* queued = createEntry.MutableQueuedCreateHandleRequest(); + *queued->MutableRequest() = createRequest; + queued->SetHandle(handle); + queued->SetNodeId(nodeId); + queued->SetRequestId(createRequest.GetHeaders().GetRequestId()); + + NProto::TQueueEntry destroyEntry; + destroyEntry.MutableDestroyHandleRequest()->SetHandle(handle); + destroyEntry.MutableDestroyHandleRequest()->SetNodeId(nodeId); + + // protobuf entries have an 8-byte header and unaligned data. This fits + // the create-confirm entry, but leaves less than one destroy entry + // free. + const ui32 queueSize = static_cast( + 8 + createEntry.ByteSizeLong() + + (8 + destroyEntry.ByteSizeLong()) / 2); + + // open fills the queue with a create confirmation. Its release + // overflows into DelayedReleaseQueue. + auto scheduler = std::make_shared(); + TBootstrap bootstrap( + CreateWallClockTimer(), scheduler, features, queueSize); + std::atomic_uint confirmCalled = 0; + std::atomic_uint destroyCalled = 0; + + bootstrap.Service->SetHandlerCreateHandle( + [=](auto, auto) + { + return MakeFuture(asyncResponse()); + }); + bootstrap.Service->SetHandlerConfirmCreateHandle( + [&](auto, auto) + { + ++confirmCalled; + return MakeFuture(NProto::TConfirmCreateHandleResponse{}); + }); + bootstrap.Service->SetHandlerDestroyHandle( + [&](auto, auto) + { + ++destroyCalled; + return MakeFuture(NProto::TDestroyHandleResponse{}); + }); + + bootstrap.Start(); + Y_DEFER { + bootstrap.Stop(); + }; + + auto moduleCounters = bootstrap.GetHandleOpsQueueCounters(); + auto entryCount = moduleCounters->FindCounter("EntryCount"); + auto overflowErrorCount = + moduleCounters->FindCounter("OverflowErrorCount"); + UNIT_ASSERT(entryCount); + UNIT_ASSERT(overflowErrorCount); + + UNIT_ASSERT_VALUES_EQUAL( + handle, + bootstrap.Fuse->SendRequest(nodeId) + .GetValue(WaitTimeout)); + bootstrap.ModuleStatsRegistry->UpdateStats(true); + UNIT_ASSERT_VALUES_EQUAL(1, entryCount->GetAtomic()); + UNIT_ASSERT_VALUES_EQUAL(0, overflowErrorCount->GetAtomic()); + UNIT_ASSERT_VALUES_EQUAL(0U, confirmCalled.load()); + + auto release = bootstrap.Fuse->SendRequest( + nodeId, + handle, + O_RDONLY); + // Destroy does not fit, so ReleaseImpl parks this FUSE request. + UNIT_ASSERT_EXCEPTION(release.GetValue(ExceptionWaitTimeout), yexception); + bootstrap.ModuleStatsRegistry->UpdateStats(true); + UNIT_ASSERT_VALUES_EQUAL(1, entryCount->GetAtomic()); + UNIT_ASSERT_VALUES_EQUAL(1, overflowErrorCount->GetAtomic()); + + // Confirm pops the create entry. + scheduler->RunAllScheduledTasks(); + UNIT_ASSERT_VALUES_EQUAL(1U, confirmCalled.load()); + UNIT_ASSERT_NO_EXCEPTION(release.GetValue(WaitTimeout)); + + // The retry queued DestroyHandle. it is processed on the next pass. + scheduler->RunAllScheduledTasks(); + UNIT_ASSERT_VALUES_EQUAL(1U, destroyCalled.load()); + } + // We want to ensure that the same file cannot be reused for FileRingBuffers // in different FileSystemLoop instances at the same time, // because it may lead to file corruption. diff --git a/cloud/filestore/libs/vfs_fuse/handle_ops_queue.cpp b/cloud/filestore/libs/vfs_fuse/handle_ops_queue.cpp index 9d20134afdc..4257738e215 100644 --- a/cloud/filestore/libs/vfs_fuse/handle_ops_queue.cpp +++ b/cloud/filestore/libs/vfs_fuse/handle_ops_queue.cpp @@ -18,6 +18,34 @@ IModuleStatsPtr THandleOpsQueue::GetModuleStats() const return Stats; } +THandleOpsQueue::EResult THandleOpsQueue::AddCreateRequest( + const NProto::TCreateHandleRequest& createHandleRequest, + ui64 nodeId, + ui64 handle, + ui64 requestId) +{ + NProto::TQueueEntry request; + auto* queued = request.MutableQueuedCreateHandleRequest(); + *queued->MutableRequest() = createHandleRequest; + queued->SetHandle(handle); + queued->SetNodeId(nodeId); + queued->SetRequestId(requestId); + + TString result; + if (!request.SerializeToString(&result)) { + Stats->IncrementSerializationErrorCount(); + return THandleOpsQueue::EResult::SerializationError; + } + + if (!RequestsToProcess.PushBack(result)) { + Stats->IncrementOverflowErrorCount(); + return THandleOpsQueue::EResult::QueueOverflow; + } + + Stats->SetEntryCount(RequestsToProcess.Size()); + return THandleOpsQueue::EResult::Ok; +} + THandleOpsQueue::EResult THandleOpsQueue::AddDestroyRequest( ui64 nodeId, ui64 handle) diff --git a/cloud/filestore/libs/vfs_fuse/handle_ops_queue.h b/cloud/filestore/libs/vfs_fuse/handle_ops_queue.h index 99dd04ef0e0..e52209bda53 100644 --- a/cloud/filestore/libs/vfs_fuse/handle_ops_queue.h +++ b/cloud/filestore/libs/vfs_fuse/handle_ops_queue.h @@ -28,7 +28,11 @@ class THandleOpsQueue explicit THandleOpsQueue(const TString& filePath, ui32 size); IModuleStatsPtr GetModuleStats() const; - + EResult AddCreateRequest( + const NProto::TCreateHandleRequest& request, + ui64 nodeId, + ui64 handle, + ui64 requestId); EResult AddDestroyRequest(ui64 nodeId, ui64 handle); std::optional Front(); void PopFront(); diff --git a/cloud/filestore/libs/vfs_fuse/loop.cpp b/cloud/filestore/libs/vfs_fuse/loop.cpp index 6b516974cf4..09992afa6fe 100644 --- a/cloud/filestore/libs/vfs_fuse/loop.cpp +++ b/cloud/filestore/libs/vfs_fuse/loop.cpp @@ -993,7 +993,8 @@ class TFileSystemLoop final THandleOpsQueuePtr handleOpsQueue; if (FileSystemConfig->GetAsyncDestroyHandleEnabled() || - FileSystemConfig->GetAsyncDestroyReadOnlyHandleEnabled()) + FileSystemConfig->GetAsyncDestroyReadOnlyHandleEnabled() || + FileSystemConfig->GetAsyncCreateHandleEnabled()) { if (Config->GetHandleOpsQueuePath()) { auto path = TFsPath(Config->GetHandleOpsQueuePath()) / @@ -1238,6 +1239,7 @@ class TFileSystemLoop final SessionState, this); + FileSystem->Init(); FuseLoop->Start(); } catch (const TServiceError& e) { error = MakeError(e.GetCode(), TString(e.GetMessage())); @@ -1295,6 +1297,8 @@ class TFileSystemLoop final features.GetAsyncDestroyHandleEnabled()); config.SetAsyncDestroyReadOnlyHandleEnabled( features.GetAsyncDestroyReadOnlyHandleEnabled()); + config.SetAsyncCreateHandleEnabled( + features.GetAsyncCreateHandleEnabled()); config.SetAsyncHandleOperationPeriod( features.GetAsyncHandleOperationPeriod()); @@ -1391,8 +1395,6 @@ class TFileSystemLoop final conn->want |= FUSE_CAP_POSIX_ACL; conn->want |= FUSE_CAP_DONT_MASK; } - - FileSystem->Init(); } void StopAsyncOnCompletionQueueStopped(TPromise stopCompleted) diff --git a/cloud/filestore/libs/vfs_fuse/protos/queue_entry.proto b/cloud/filestore/libs/vfs_fuse/protos/queue_entry.proto index f4fe86dafc7..ae1113a6f7e 100644 --- a/cloud/filestore/libs/vfs_fuse/protos/queue_entry.proto +++ b/cloud/filestore/libs/vfs_fuse/protos/queue_entry.proto @@ -4,10 +4,19 @@ import "cloud/filestore/public/api/protos/data.proto"; package NCloud.NFileStore.NProto; +message TQueuedCreateHandleRequest +{ + TCreateHandleRequest Request = 1; + uint64 Handle = 2; + uint64 NodeId = 3; + uint64 RequestId = 4; +} + message TQueueEntry { oneof Request { TCreateHandleRequest CreateHandleRequest = 1; TDestroyHandleRequest DestroyHandleRequest = 2; + TQueuedCreateHandleRequest QueuedCreateHandleRequest = 3; } } diff --git a/cloud/filestore/public/api/grpc/service.proto b/cloud/filestore/public/api/grpc/service.proto index bdc790aa9df..fae42338e35 100644 --- a/cloud/filestore/public/api/grpc/service.proto +++ b/cloud/filestore/public/api/grpc/service.proto @@ -319,6 +319,13 @@ service TFileStoreService }; } + rpc ConfirmCreateHandle(TConfirmCreateHandleRequest) returns (TConfirmCreateHandleResponse) { + option (google.api.http) = { + post: "/confirm_create_handle" + body: "*" + }; + } + rpc DestroyHandle(TDestroyHandleRequest) returns (TDestroyHandleResponse) { option (google.api.http) = { post: "/destroy_handle" diff --git a/cloud/filestore/public/api/protos/data.proto b/cloud/filestore/public/api/protos/data.proto index 559d1b356a1..926cf26cf82 100644 --- a/cloud/filestore/public/api/protos/data.proto +++ b/cloud/filestore/public/api/protos/data.proto @@ -67,6 +67,10 @@ message TCreateHandleRequest // Umask of the calling process uint32 Umask = 13; + + // If set, the client allows CreateHandle to return before the handle is + // durably persisted. + bool AllowAsyncCreateHandle = 14; } message TCreateHandleResponse @@ -90,6 +94,40 @@ message TCreateHandleResponse // kernel this refers to the page cache bool GuestKeepCache = 6; + // If set, the handle was returned before durable persistence completed and + // the client must confirm it. + bool HandleCreatedAsync = 7; + + // Optional response headers. + TResponseHeaders Headers = 1000; +} + +message TConfirmCreateHandleRequest +{ + // Optional request headers. + THeaders Headers = 1; + + // FileSystem identifier. + bytes FileSystemId = 2; + + // Node. + uint64 NodeId = 3; + + // IO handle. + uint64 Handle = 4; + + // Original CreateHandle flags. + uint32 Flags = 5; + + // Original CreateHandle request id. + uint64 RequestId = 6; +} + +message TConfirmCreateHandleResponse +{ + // Optional error, set only if error happened. + NCloud.NProto.TError Error = 1; + // Optional response headers. TResponseHeaders Headers = 1000; } diff --git a/cloud/filestore/public/api/protos/fs.proto b/cloud/filestore/public/api/protos/fs.proto index 5bc5886debd..0f21e672695 100644 --- a/cloud/filestore/public/api/protos/fs.proto +++ b/cloud/filestore/public/api/protos/fs.proto @@ -63,6 +63,7 @@ message TFileStoreFeatures bool XAttrCacheInvalidateOnCreateEnabled = 48; bool AsyncDestroyReadOnlyHandleEnabled = 49; bool ExternalWriteDataPayloadEnabled = 50; + bool AsyncCreateHandleEnabled = 51; } message TFileStore diff --git a/cloud/filestore/tools/analytics/libs/event-log/dump_ut.cpp b/cloud/filestore/tools/analytics/libs/event-log/dump_ut.cpp index 64b093163c6..8a4d963c2ce 100644 --- a/cloud/filestore/tools/analytics/libs/event-log/dump_ut.cpp +++ b/cloud/filestore/tools/analytics/libs/event-log/dump_ut.cpp @@ -94,7 +94,7 @@ Y_UNIT_TEST_SUITE(TDumpTest) { const auto requests = GetRequestTypes(); - UNIT_ASSERT_VALUES_EQUAL(80, requests.size()); + UNIT_ASSERT_VALUES_EQUAL(81, requests.size()); ui32 index = 0; #define TEST_REQUEST_TYPE(id, name) \ @@ -151,20 +151,21 @@ Y_UNIT_TEST_SUITE(TDumpTest) TEST_REQUEST_TYPE(43, ReadData); TEST_REQUEST_TYPE(44, WriteData); TEST_REQUEST_TYPE(45, AllocateData); - TEST_REQUEST_TYPE(46, Fsync); - TEST_REQUEST_TYPE(47, FsyncDir); - TEST_REQUEST_TYPE(48, GetSessionEventsStream); - TEST_REQUEST_TYPE(49, StartEndpoint); - TEST_REQUEST_TYPE(50, StopEndpoint); - TEST_REQUEST_TYPE(51, ListEndpoints); - TEST_REQUEST_TYPE(52, KickEndpoint); - TEST_REQUEST_TYPE(53, DescribeData); - TEST_REQUEST_TYPE(54, GenerateBlobIds); - TEST_REQUEST_TYPE(55, AddData); - TEST_REQUEST_TYPE(56, ReadBlob); - TEST_REQUEST_TYPE(57, WriteBlob); - TEST_REQUEST_TYPE(58, ConfirmAddData); - TEST_REQUEST_TYPE(59, CancelAddData); + TEST_REQUEST_TYPE(46, ConfirmCreateHandle); + TEST_REQUEST_TYPE(47, Fsync); + TEST_REQUEST_TYPE(48, FsyncDir); + TEST_REQUEST_TYPE(49, GetSessionEventsStream); + TEST_REQUEST_TYPE(50, StartEndpoint); + TEST_REQUEST_TYPE(51, StopEndpoint); + TEST_REQUEST_TYPE(52, ListEndpoints); + TEST_REQUEST_TYPE(53, KickEndpoint); + TEST_REQUEST_TYPE(54, DescribeData); + TEST_REQUEST_TYPE(55, GenerateBlobIds); + TEST_REQUEST_TYPE(56, AddData); + TEST_REQUEST_TYPE(57, ReadBlob); + TEST_REQUEST_TYPE(58, WriteBlob); + TEST_REQUEST_TYPE(59, ConfirmAddData); + TEST_REQUEST_TYPE(60, CancelAddData); // Fuse TEST_REQUEST_TYPE(1001, Flush); From 520c0189ab8c8ea00219a79a7f9ed3bd6fe3679c Mon Sep 17 00:00:00 2001 From: Evgeny Budilovsky Date: Thu, 23 Jul 2026 09:31:13 +0000 Subject: [PATCH 2/2] [Filestore] issue-5432: add basic async open support (tablet) --- cloud/filestore/libs/storage/api/service.h | 5 +- cloud/filestore/libs/storage/core/config.cpp | 1 + cloud/filestore/libs/storage/core/config.h | 1 + .../storage/service/service_actor_forward.cpp | 2 +- .../tablet/tablet_actor_createhandle.cpp | 216 +++++- .../tablet/tablet_actor_createsession.cpp | 2 + .../libs/storage/tablet/tablet_counters.h | 1 + .../filestore/libs/storage/tablet/tablet_tx.h | 37 + .../libs/storage/tablet/tablet_ut_handles.cpp | 630 ++++++++++++++++++ .../libs/storage/testlib/tablet_client.h | 15 + 10 files changed, 895 insertions(+), 15 deletions(-) diff --git a/cloud/filestore/libs/storage/api/service.h b/cloud/filestore/libs/storage/api/service.h index 312c5f2bf62..0e3e6b14c07 100644 --- a/cloud/filestore/libs/storage/api/service.h +++ b/cloud/filestore/libs/storage/api/service.h @@ -45,6 +45,7 @@ namespace NCloud::NFileStore::NStorage { // FILESTORE_SERVICE_REQUESTS_FWD_TO_SHARD_BY_NODE_ID #define FILESTORE_SERVICE_REQUESTS_FWD_TO_SHARD_BY_HANDLE(xxx, ...) \ + xxx(ConfirmCreateHandle, __VA_ARGS__) \ xxx(DestroyHandle, __VA_ARGS__) \ xxx(AllocateData, __VA_ARGS__) \ \ @@ -258,8 +259,8 @@ struct TEvService EvRegisterLocalFileStore = EvBegin + 71, EvUnregisterLocalFileStore, - // EvGetSessionAttrRequest - EvUnused2 = EvBegin + 73, + EvConfirmCreateHandleRequest = EvBegin + 73, + EvConfirmCreateHandleResponse, EvAddClusterNodeRequest = EvBegin + 75, EvAddClusterNodeResponse, diff --git a/cloud/filestore/libs/storage/core/config.cpp b/cloud/filestore/libs/storage/core/config.cpp index d2a59fe80e4..3e5a926c4c7 100644 --- a/cloud/filestore/libs/storage/core/config.cpp +++ b/cloud/filestore/libs/storage/core/config.cpp @@ -260,6 +260,7 @@ using TAliases = NProto::TStorageConfig::TFilestoreAliases; \ xxx(AsyncDestroyHandleEnabled, bool, false )\ xxx(AsyncDestroyReadOnlyHandleEnabled, bool, false )\ + xxx(AsyncCreateHandleEnabled, bool, false )\ xxx(TabletUnsafeAsyncReadOnlyCreateHandleEnabled, bool, false )\ xxx(TabletUnsafeAsyncDestroyHandleEnabled, bool, false )\ xxx(AsyncHandleOperationPeriod, TDuration, TDuration::MilliSeconds(50))\ diff --git a/cloud/filestore/libs/storage/core/config.h b/cloud/filestore/libs/storage/core/config.h index 6b7760b076f..89204454170 100644 --- a/cloud/filestore/libs/storage/core/config.h +++ b/cloud/filestore/libs/storage/core/config.h @@ -273,6 +273,7 @@ class TStorageConfig bool GetAsyncDestroyHandleEnabled() const; bool GetAsyncDestroyReadOnlyHandleEnabled() const; + bool GetAsyncCreateHandleEnabled() const; bool GetTabletUnsafeAsyncReadOnlyCreateHandleEnabled() const; bool GetTabletUnsafeAsyncDestroyHandleEnabled() const; TDuration GetAsyncHandleOperationPeriod() const; diff --git a/cloud/filestore/libs/storage/service/service_actor_forward.cpp b/cloud/filestore/libs/storage/service/service_actor_forward.cpp index dbf0ae24d31..9eeb894be23 100644 --- a/cloud/filestore/libs/storage/service/service_actor_forward.cpp +++ b/cloud/filestore/libs/storage/service/service_actor_forward.cpp @@ -321,7 +321,7 @@ void TStorageServiceActor::ForwardRequestToShard( FILESTORE_FORWARD_REQUEST_TO_SHARD_BY_HANDLE, TEvService) -#undef FILESTORE_FORWARD_REQUEST_TO_SHARD_BY_NODE_ID +#undef FILESTORE_FORWARD_REQUEST_TO_SHARD_BY_HANDLE #define FILESTORE_DEFINE_HANDLE_FORWARD(name, ns) \ template void TStorageServiceActor::ForwardRequest( \ diff --git a/cloud/filestore/libs/storage/tablet/tablet_actor_createhandle.cpp b/cloud/filestore/libs/storage/tablet/tablet_actor_createhandle.cpp index 8766a672c05..09b897c62b8 100644 --- a/cloud/filestore/libs/storage/tablet/tablet_actor_createhandle.cpp +++ b/cloud/filestore/libs/storage/tablet/tablet_actor_createhandle.cpp @@ -18,7 +18,8 @@ namespace { //////////////////////////////////////////////////////////////////////////////// -NProto::TError ValidateRequest(const NProto::TCreateHandleRequest& request) +NProto::TError ValidateCreateHandleRequest( + const NProto::TCreateHandleRequest& request) { if (request.GetNodeId() == InvalidNodeId) { return ErrorInvalidArgument(); @@ -41,6 +42,16 @@ NProto::TError ValidateRequest(const NProto::TCreateHandleRequest& request) return {}; } +NProto::TError ValidateConfirmCreateHandleRequest( + const NProto::TConfirmCreateHandleRequest& request) +{ + if (request.GetNodeId() == InvalidNodeId || !request.GetHandle()) { + return ErrorInvalidArgument(); + } + + return {}; +} + } // namespace //////////////////////////////////////////////////////////////////////////////// @@ -51,8 +62,10 @@ void TIndexTabletActor::HandleCreateHandle( { using TResponse = TEvService::TEvCreateHandleResponse; - auto* session = - AcceptRequest(ev, ctx, ValidateRequest); + auto* session = AcceptRequest( + ev, + ctx, + ValidateCreateHandleRequest); if (!session) { return; } @@ -513,6 +526,29 @@ void TIndexTabletActor::ExecuteTx_CreateHandle( WriteOpLogEntry(*db, args.OpLogEntry); } + constexpr ui32 wflags = (ProtoFlag(NProto::TCreateHandleRequest::E_CREATE) + | ProtoFlag(NProto::TCreateHandleRequest::E_WRITE) + | ProtoFlag(NProto::TCreateHandleRequest::E_APPEND) + | ProtoFlag(NProto::TCreateHandleRequest::E_TRUNCATE)); + + const bool isReadOnly = (args.Flags & wflags) == 0; + + const bool hasLocalHandle = args.Response.GetHandle() != 0; + + const bool safeAsync = hasLocalHandle && + isReadOnly && + Config->GetAsyncCreateHandleEnabled() && + args.Request.GetAllowAsyncCreateHandle(); + + const bool unsafeAsync = + hasLocalHandle && + isReadOnly && + Config->GetTabletUnsafeAsyncReadOnlyCreateHandleEnabled(); + + if (safeAsync) { + args.Response.SetHandleCreatedAsync(true); + } + AddDupCacheEntry( *db, session, @@ -522,15 +558,7 @@ void TIndexTabletActor::ExecuteTx_CreateHandle( EnqueueTruncateIfNeeded(ctx); - constexpr ui32 wflags = (ProtoFlag(NProto::TCreateHandleRequest::E_CREATE) - | ProtoFlag(NProto::TCreateHandleRequest::E_WRITE) - | ProtoFlag(NProto::TCreateHandleRequest::E_APPEND) - | ProtoFlag(NProto::TCreateHandleRequest::E_TRUNCATE)); - - const bool isReadOnly = (args.Flags & wflags) == 0; - - if (Config->GetTabletUnsafeAsyncReadOnlyCreateHandleEnabled() && isReadOnly) - { + if (safeAsync || unsafeAsync) { LOG_DEBUG( ctx, TFileStoreComponents::TABLET, @@ -629,4 +657,168 @@ void TIndexTabletActor::CompleteTx_CreateHandle( CompleteCreateHandle(ctx, args); } +//////////////////////////////////////////////////////////////////////////////// + +void TIndexTabletActor::HandleConfirmCreateHandle( + const TEvService::TEvConfirmCreateHandleRequest::TPtr& ev, + const TActorContext& ctx) +{ + auto* session = AcceptRequest( + ev, + ctx, + ValidateConfirmCreateHandleRequest); + if (!session) { + return; + } + + auto* msg = ev->Get(); + auto requestInfo = CreateRequestInfo( + ev->Sender, + ev->Cookie, + msg->CallContext); + requestInfo->StartedTs = ctx.Now(); + + AddInFlightRequest(*requestInfo); + + ExecuteTx( + ctx, + std::move(requestInfo), + std::move(msg->Record)); +} + +bool TIndexTabletActor::PrepareTx_ConfirmCreateHandle( + const TActorContext& ctx, + TTransactionContext& tx, + TTxIndexTablet::TConfirmCreateHandle& args) +{ + Y_UNUSED(ctx); + + FILESTORE_VALIDATE_TX_SESSION(ConfirmCreateHandle, args); + + auto* session = FindSession( + args.ClientId, + args.SessionId, + args.SessionSeqNo); + if (!session) { + auto message = ReportSessionNotFoundInTx(TStringBuilder() + << "ConfirmCreateHandle: " << args.Request.ShortDebugString()); + args.Error = MakeError(E_INVALID_STATE, std::move(message)); + return true; + } + + args.ReadCommitId = GetReadCommitId(session->GetCheckpointId()); + if (args.ReadCommitId == InvalidCommitId) { + args.Error = ErrorInvalidCheckpoint(session->GetCheckpointId()); + return true; + } + + TIndexTabletDatabase db(tx.DB); + if (!ReadNode(db, args.NodeId, args.ReadCommitId, args.Node)) { + return false; + } + + if (!args.Node) { + args.Error = ErrorInvalidTarget(args.NodeId); + return true; + } + + if (args.Node->Attrs.GetType() != NProto::E_REGULAR_NODE) { + args.Error = ErrorIsDirectory(args.NodeId); + return true; + } + + return true; +} + +void TIndexTabletActor::ExecuteTx_ConfirmCreateHandle( + const TActorContext& ctx, + TTransactionContext& tx, + TTxIndexTablet::TConfirmCreateHandle& args) +{ + Y_UNUSED(ctx); + + FILESTORE_VALIDATE_TX_ERROR(ConfirmCreateHandle, args); + + auto* session = FindSession(args.SessionId); + if (!session) { + auto message = ReportSessionNotFoundInTx(TStringBuilder() + << "ConfirmCreateHandle: " << args.Request.ShortDebugString()); + args.Error = MakeError(E_INVALID_STATE, std::move(message)); + return; + } + + TIndexTabletDatabase db(tx.DB); + + if (auto* handle = FindHandle(args.Handle)) { + if (handle->Session != session || handle->GetNodeId() != args.NodeId) { + auto message = TStringBuilder() + << "ConfirmCreateHandle collision: " + << args.Request.ShortDebugString() + << " existing session: " << handle->GetSessionId() + << " existing node: " << handle->GetNodeId(); + args.Error = MakeError(E_INVALID_STATE, std::move(message)); + return; + } + } else { + handle = UnsafeCreateHandle( + db, + session, + args.Handle, + args.NodeId, + session->GetCheckpointId() ? args.ReadCommitId : InvalidCommitId, + args.Flags); + + if (!handle) { + auto message = ReportFailedToCreateHandle(TStringBuilder() + << "ConfirmCreateHandle: " << args.Request.ShortDebugString()); + args.Error = MakeError(E_INVALID_STATE, std::move(message)); + return; + } + } + + if (args.CreateRequestId && !session->LookupDupEntry(args.CreateRequestId)) + { + NProto::TCreateHandleResponse response; + response.SetHandle(args.Handle); + ConvertNodeFromAttrs( + *response.MutableNodeAttr(), + args.NodeId, + args.Node->Attrs); + + AddDupCacheEntry( + db, + session, + args.CreateRequestId, + response, + Config->GetDupCacheEntryCount()); + } +} + +void TIndexTabletActor::CompleteTx_ConfirmCreateHandle( + const TActorContext& ctx, + TTxIndexTablet::TConfirmCreateHandle& args) +{ + RemoveInFlightRequest(*args.RequestInfo); + + auto response = + std::make_unique( + args.Error); + + if (!HasError(args.Error)) { + CommitDupCacheEntry(args.SessionId, args.CreateRequestId); + } + + Metrics.ConfirmCreateHandle.Update( + 1, + 0, + ctx.Now() - args.RequestInfo->StartedTs); + + CompleteResponse( + response->Record, + args.RequestInfo->CallContext, + ctx); + + NCloud::Reply(ctx, *args.RequestInfo, std::move(response)); +} + } // namespace NCloud::NFileStore::NStorage diff --git a/cloud/filestore/libs/storage/tablet/tablet_actor_createsession.cpp b/cloud/filestore/libs/storage/tablet/tablet_actor_createsession.cpp index 74040f2a2aa..1210155db72 100644 --- a/cloud/filestore/libs/storage/tablet/tablet_actor_createsession.cpp +++ b/cloud/filestore/libs/storage/tablet/tablet_actor_createsession.cpp @@ -50,6 +50,8 @@ void FillFeatures( config.GetAsyncDestroyHandleEnabled()); features->SetAsyncDestroyReadOnlyHandleEnabled( config.GetAsyncDestroyReadOnlyHandleEnabled()); + features->SetAsyncCreateHandleEnabled( + config.GetAsyncCreateHandleEnabled()); features->SetAsyncHandleOperationPeriod( config.GetAsyncHandleOperationPeriod().MilliSeconds()); diff --git a/cloud/filestore/libs/storage/tablet/tablet_counters.h b/cloud/filestore/libs/storage/tablet/tablet_counters.h index 248bed273d0..6545565e032 100644 --- a/cloud/filestore/libs/storage/tablet/tablet_counters.h +++ b/cloud/filestore/libs/storage/tablet/tablet_counters.h @@ -94,6 +94,7 @@ TTabletCountersPtr CreateIndexTabletCounters(); xxx(GetNodeAttrInShard, __VA_ARGS__) \ xxx(CreateHandle, __VA_ARGS__) \ xxx(CreateHandleInShard, __VA_ARGS__) \ + xxx(ConfirmCreateHandle, __VA_ARGS__) \ xxx(DestroyHandle, __VA_ARGS__) \ xxx(CreateNode, __VA_ARGS__) \ xxx(CreateNodeInShard, __VA_ARGS__) \ diff --git a/cloud/filestore/libs/storage/tablet/tablet_tx.h b/cloud/filestore/libs/storage/tablet/tablet_tx.h index a3dbd3e123b..667153d656a 100644 --- a/cloud/filestore/libs/storage/tablet/tablet_tx.h +++ b/cloud/filestore/libs/storage/tablet/tablet_tx.h @@ -124,6 +124,7 @@ namespace NCloud::NFileStore::NStorage { xxx(SetHasXAttrs, __VA_ARGS__) \ \ xxx(CreateHandle, __VA_ARGS__) \ + xxx(ConfirmCreateHandle, __VA_ARGS__) \ xxx(DestroyHandle, __VA_ARGS__) \ \ xxx(AcquireLock, __VA_ARGS__) \ @@ -1905,6 +1906,42 @@ struct TTxIndexTablet } }; + struct TConfirmCreateHandle + : TTxIndexTabletBase + , TErrorAware + , TSessionAware + { + const TRequestInfoPtr RequestInfo; + const NProto::TConfirmCreateHandleRequest Request; + const ui64 NodeId; + const ui64 Handle; + const ui32 Flags; + const ui64 CreateRequestId; + + ui64 ReadCommitId = InvalidCommitId; + TMaybe Node; + + TConfirmCreateHandle( + TRequestInfoPtr requestInfo, + NProto::TConfirmCreateHandleRequest request) + : TSessionAware(request) + , RequestInfo(std::move(requestInfo)) + , Request(std::move(request)) + , NodeId(Request.GetNodeId()) + , Handle(Request.GetHandle()) + , Flags(Request.GetFlags()) + , CreateRequestId(Request.GetRequestId()) + {} + + void Clear() override + { + TErrorAware::Clear(); + + ReadCommitId = InvalidCommitId; + Node.Clear(); + } + }; + // // DestroyHandle // diff --git a/cloud/filestore/libs/storage/tablet/tablet_ut_handles.cpp b/cloud/filestore/libs/storage/tablet/tablet_ut_handles.cpp index 84e7a37bf82..1fb576377a6 100644 --- a/cloud/filestore/libs/storage/tablet/tablet_ut_handles.cpp +++ b/cloud/filestore/libs/storage/tablet/tablet_ut_handles.cpp @@ -109,6 +109,636 @@ Y_UNIT_TEST_SUITE(TIndexTabletTest_Handles) } } + Y_UNIT_TEST(ShouldSetHandleCreatedAsyncForEligibleCreateHandle) + { + NProto::TStorageConfig storageConfig; + storageConfig.SetAsyncCreateHandleEnabled(true); + TTestEnv env({}, storageConfig); + + ui32 nodeIdx = env.AddDynamicNode(); + ui64 tabletId = env.BootIndexTablet(nodeIdx); + + TIndexTabletClient tablet(env.GetRuntime(), nodeIdx, tabletId); + tablet.InitSession("client", "session"); + + auto id = CreateNode(tablet, TCreateNodeArgs::File(RootNodeId, "test")); + + { + // Read-only open with client opt-in may be acknowledged before the + // handle is durably persisted. + auto request = + tablet.CreateCreateHandleRequest(id, TCreateHandleArgs::RDNLY); + request->Record.SetAllowAsyncCreateHandle(true); + tablet.SendRequest(std::move(request)); + + auto response = tablet.RecvCreateHandleResponse(); + UNIT_ASSERT_VALUES_EQUAL_C( + S_OK, + response->Record.GetError().GetCode(), + response->Record.GetError().GetMessage()); + UNIT_ASSERT(response->Record.GetHandleCreatedAsync()); + } + + { + // Without client opt-in, the tablet must use the synchronous path + // and must not mark the handle as created asynchronously. + auto request = + tablet.CreateCreateHandleRequest(id, TCreateHandleArgs::RDNLY); + tablet.SendRequest(std::move(request)); + + auto response = tablet.RecvCreateHandleResponse(); + UNIT_ASSERT_VALUES_EQUAL_C( + S_OK, + response->Record.GetError().GetCode(), + response->Record.GetError().GetMessage()); + UNIT_ASSERT(!response->Record.GetHandleCreatedAsync()); + } + + { + // Write opens are ineligible even when the client sets the async + // opt-in bit. + auto request = + tablet.CreateCreateHandleRequest(id, TCreateHandleArgs::WRNLY); + request->Record.SetAllowAsyncCreateHandle(true); + tablet.SendRequest(std::move(request)); + + auto response = tablet.RecvCreateHandleResponse(); + UNIT_ASSERT_VALUES_EQUAL_C( + S_OK, + response->Record.GetError().GetCode(), + response->Record.GetError().GetMessage()); + UNIT_ASSERT(!response->Record.GetHandleCreatedAsync()); + } + } + + Y_UNIT_TEST(ShouldCacheAsyncCreateHandleFlagForDuplicateRequests) + { + NProto::TStorageConfig storageConfig; + storageConfig.SetAsyncCreateHandleEnabled(true); + TTestEnv env({}, storageConfig); + + ui32 nodeIdx = env.AddDynamicNode(); + ui64 tabletId = env.BootIndexTablet(nodeIdx); + + TIndexTabletClient tablet(env.GetRuntime(), nodeIdx, tabletId); + tablet.InitSession("client", "session"); + + auto id = CreateNode(tablet, TCreateNodeArgs::File(RootNodeId, "test")); + constexpr ui64 requestId = 100500; + + auto createRequest = [&] + { + auto request = tablet.CreateCreateHandleRequest( + id, + TCreateHandleArgs::RDNLY); + request->Record.SetAllowAsyncCreateHandle(true); + request->Record.MutableHeaders()->SetRequestId(requestId); + return request; + }; + + tablet.SendRequest(createRequest()); + auto response = tablet.RecvCreateHandleResponse(); + UNIT_ASSERT(response->Record.GetHandleCreatedAsync()); + const ui64 handle = response->Record.GetHandle(); + + // A retry is served from the duplicate cache and must still require + // confirmation of the handle. + tablet.SendRequest(createRequest()); + response = tablet.RecvCreateHandleResponse(); + UNIT_ASSERT(response->Record.GetHandleCreatedAsync()); + UNIT_ASSERT_VALUES_EQUAL(handle, response->Record.GetHandle()); + + tablet.ConfirmCreateHandle( + id, + handle, + TCreateHandleArgs::RDNLY, + requestId); + tablet.DestroyHandle(handle); + } + + Y_UNIT_TEST(ShouldIgnoreAsyncCreateHandleOptInIfFeatureDisabled) + { + NProto::TStorageConfig storageConfig; + storageConfig.SetAsyncCreateHandleEnabled(false); + TTestEnv env({}, storageConfig); + + ui32 nodeIdx = env.AddDynamicNode(); + ui64 tabletId = env.BootIndexTablet(nodeIdx); + + TIndexTabletClient tablet(env.GetRuntime(), nodeIdx, tabletId); + tablet.InitSession("client", "session"); + + auto id = CreateNode(tablet, TCreateNodeArgs::File(RootNodeId, "test")); + + // Hold the create tx commit so the async response is acknowledged + // before the SessionHandles row becomes durable. + TAutoPtr putEvent; + env.GetRuntime().SetEventFilter( + [&](auto& runtime, auto& ev) + { + Y_UNUSED(runtime); + if (!putEvent && + ev->GetTypeRewrite() == TEvBlobStorage::EvPut) + { + putEvent = std::move(ev); + return true; + } + + return false; + }); + + auto request = + tablet.CreateCreateHandleRequest(id, TCreateHandleArgs::RDNLY); + request->Record.SetAllowAsyncCreateHandle(true); + tablet.SendRequest(std::move(request)); + + env.GetRuntime().DispatchEvents(TDispatchOptions{ + .CustomFinalCondition = [&]() + { + return putEvent != nullptr; + }}); + + tablet.AssertCreateHandleNoResponse(); + + env.GetRuntime().SetEventFilter( + TTestActorRuntimeBase::DefaultFilterFunc); + env.GetRuntime().Send(putEvent.Release(), nodeIdx); + + auto response = tablet.RecvCreateHandleResponse(); + UNIT_ASSERT_VALUES_EQUAL_C( + S_OK, + response->Record.GetError().GetCode(), + response->Record.GetError().GetMessage()); + UNIT_ASSERT(!response->Record.GetHandleCreatedAsync()); + UNIT_ASSERT(response->Record.GetHandle()); + } + + Y_UNIT_TEST(ShouldNotMarkShardRedirectCreateHandleAsAsync) + { + NProto::TStorageConfig storageConfig; + storageConfig.SetAsyncCreateHandleEnabled(true); + TTestEnv env({}, storageConfig); + + ui32 nodeIdx = env.AddDynamicNode(); + ui64 tabletId = env.BootIndexTablet(nodeIdx); + + TIndexTabletClient tablet(env.GetRuntime(), nodeIdx, tabletId); + tablet.InitSession("client", "session"); + + const TString shardId = "shard"; + const TString name = "test"; + const TString shardNodeName = CreateGuidAsString(); + // Model an existing namespace entry that lives in a shard. Opening it + // on the main tablet should return a redirect, not a local handle. + CreateExternalRef(tablet, RootNodeId, name, shardId, shardNodeName); + + // Even with async opt-in, a redirect response has no handle to confirm. + auto request = tablet.CreateCreateHandleRequest( + RootNodeId, + name, + TCreateHandleArgs::RDNLY); + request->Record.SetAllowAsyncCreateHandle(true); + tablet.SendRequest(std::move(request)); + + auto response = tablet.RecvCreateHandleResponse(); + UNIT_ASSERT_VALUES_EQUAL_C( + S_OK, + response->Record.GetError().GetCode(), + response->Record.GetError().GetMessage()); + + // HandleCreatedAsync is meaningful only for a real local handle. + UNIT_ASSERT_VALUES_EQUAL(0, response->Record.GetHandle()); + UNIT_ASSERT(!response->Record.GetHandleCreatedAsync()); + UNIT_ASSERT_VALUES_EQUAL( + shardId, + response->Record.GetShardFileSystemId()); + UNIT_ASSERT_VALUES_EQUAL( + shardNodeName, + response->Record.GetShardNodeName()); + } + + Y_UNIT_TEST(ShouldRecreateExactHandleOnConfirmCreateHandle) + { + TTestEnv env; + + ui32 nodeIdx = env.AddDynamicNode(); + ui64 tabletId = env.BootIndexTablet(nodeIdx); + + TIndexTabletClient tablet(env.GetRuntime(), nodeIdx, tabletId); + tablet.InitSession("client", "session"); + + auto id = CreateNode(tablet, TCreateNodeArgs::File(RootNodeId, "test")); + const ui64 handle = 424242; + + tablet.ConfirmCreateHandle( + id, + handle, + TCreateHandleArgs::RDNLY, + 100500); + + tablet.DescribeData(handle, 0, 1_KB); + tablet.DestroyHandle(handle); + } + + Y_UNIT_TEST(ShouldConfirmCreateHandleIdempotently) + { + TTestEnv env; + + ui32 nodeIdx = env.AddDynamicNode(); + ui64 tabletId = env.BootIndexTablet(nodeIdx); + + TIndexTabletClient tablet(env.GetRuntime(), nodeIdx, tabletId); + tablet.InitSession("client", "session"); + + auto id = CreateNode(tablet, TCreateNodeArgs::File(RootNodeId, "test")); + const ui64 handle = 424243; + + tablet.ConfirmCreateHandle( + id, + handle, + TCreateHandleArgs::RDNLY, + 100500); + tablet.ConfirmCreateHandle( + id, + handle, + TCreateHandleArgs::RDNLY, + 100500); + + tablet.DescribeData(handle, 0, 1_KB); + tablet.DestroyHandle(handle); + } + + Y_UNIT_TEST(ShouldConfirmAsyncCreateHandleAfterTabletRestart) + { + NProto::TStorageConfig storageConfig; + storageConfig.SetAsyncCreateHandleEnabled(true); + TTestEnv env({}, storageConfig); + + ui32 nodeIdx = env.AddDynamicNode(); + ui64 tabletId = env.BootIndexTablet(nodeIdx); + + TIndexTabletClient tablet(env.GetRuntime(), nodeIdx, tabletId); + tablet.InitSession("client", "session"); + + auto id = CreateNode(tablet, TCreateNodeArgs::File(RootNodeId, "test")); + + // Hold the create tx commit so the async response is acknowledged + // before the SessionHandles row becomes durable. + TAutoPtr putEvent; + env.GetRuntime().SetEventFilter( + [&](auto& runtime, auto& ev) + { + Y_UNUSED(runtime); + if (!putEvent && + ev->GetTypeRewrite() == TEvBlobStorage::EvPut) + { + putEvent = std::move(ev); + return true; + } + + return false; + }); + + auto request = + tablet.CreateCreateHandleRequest(id, TCreateHandleArgs::RDNLY); + request->Record.SetAllowAsyncCreateHandle(true); + request->Record.MutableHeaders()->SetRequestId(100500); + tablet.SendRequest(std::move(request)); + + env.GetRuntime().DispatchEvents(TDispatchOptions{ + .CustomFinalCondition = [&]() + { + return putEvent != nullptr; + }}); + + auto response = tablet.RecvCreateHandleResponse(); + UNIT_ASSERT_VALUES_EQUAL_C( + S_OK, + response->Record.GetError().GetCode(), + response->Record.GetError().GetMessage()); + UNIT_ASSERT(response->Record.GetHandleCreatedAsync()); + + const ui64 handle = response->Record.GetHandle(); + UNIT_ASSERT(handle); + + // Drop the uncommitted tx by rebooting the tablet before releasing the + // captured TEvPut. + env.GetRuntime().SetEventFilter( + TTestActorRuntimeBase::DefaultFilterFunc); + tablet.RebootTablet(); + tablet.RecoverSession(); + + // The early-returned handle is not persisted yet, so recovery must + // explicitly confirm that exact handle id. + auto describeResponse = tablet.AssertDescribeDataFailed( + handle, + 0, + 1_KB); + UNIT_ASSERT_VALUES_EQUAL_C( + E_FS_BADHANDLE, + describeResponse->Record.GetError().GetCode(), + describeResponse->Record.GetError().GetMessage()); + + tablet.ConfirmCreateHandle( + id, + handle, + TCreateHandleArgs::RDNLY, + 100500); + + // ConfirmCreateHandle recreates and persists the exact returned handle. + tablet.DescribeData(handle, 0, 1_KB); + tablet.DestroyHandle(handle); + } + + Y_UNIT_TEST(ShouldRestoreCreateHandleDupCacheAfterPostRestartConfirm) + { + NProto::TStorageConfig storageConfig; + storageConfig.SetAsyncCreateHandleEnabled(true); + TTestEnv env({}, storageConfig); + + ui32 nodeIdx = env.AddDynamicNode(); + ui64 tabletId = env.BootIndexTablet(nodeIdx); + + TIndexTabletClient tablet(env.GetRuntime(), nodeIdx, tabletId); + tablet.InitSession("client", "session"); + + auto node = + CreateNode(tablet, TCreateNodeArgs::File(RootNodeId, "test")); + constexpr ui64 requestId = 100500; + + auto createRequest = [&] + { + auto request = tablet.CreateCreateHandleRequest( + node, + TCreateHandleArgs::RDNLY); + request->Record.SetAllowAsyncCreateHandle(true); + request->Record.MutableHeaders()->SetRequestId(requestId); + return request; + }; + + // Hold the create tx commit so both the handle and its dup-cache entry + // are lost when the tablet restarts. + TAutoPtr putEvent; + env.GetRuntime().SetEventFilter( + [&](auto& runtime, auto& ev) + { + Y_UNUSED(runtime); + if (!putEvent && + ev->GetTypeRewrite() == TEvBlobStorage::EvPut) + { + putEvent = std::move(ev); + return true; + } + + return false; + }); + + tablet.SendRequest(createRequest()); + + env.GetRuntime().DispatchEvents(TDispatchOptions{ + .CustomFinalCondition = [&]() + { + return putEvent != nullptr; + }}); + + auto response = tablet.RecvCreateHandleResponse(); + UNIT_ASSERT_VALUES_EQUAL_C( + S_OK, + response->Record.GetError().GetCode(), + response->Record.GetError().GetMessage()); + UNIT_ASSERT(response->Record.GetHandleCreatedAsync()); + + const ui64 confirmedHandle = response->Record.GetHandle(); + UNIT_ASSERT(confirmedHandle); + + env.GetRuntime().SetEventFilter( + TTestActorRuntimeBase::DefaultFilterFunc); + tablet.RebootTablet(); + tablet.RecoverSession(); + + tablet.ConfirmCreateHandle( + node, + confirmedHandle, + TCreateHandleArgs::RDNLY, + requestId); + tablet.DescribeData(confirmedHandle, 0, 1_KB); + + // ConfirmCreateHandle restores the lost CreateHandle dup-cache entry. + // Retrying the original CreateHandle request id returns the same + // confirmed handle instead of opening another one. + tablet.SendRequest(createRequest()); + response = tablet.RecvCreateHandleResponse(); + UNIT_ASSERT_VALUES_EQUAL_C( + S_OK, + response->Record.GetError().GetCode(), + response->Record.GetError().GetMessage()); + UNIT_ASSERT(!response->Record.GetHandleCreatedAsync()); + + const ui64 rerunHandle = response->Record.GetHandle(); + UNIT_ASSERT_VALUES_EQUAL(confirmedHandle, rerunHandle); + + tablet.DestroyHandle(confirmedHandle); + } + + Y_UNIT_TEST(ShouldDelayAsyncCreateHandleConfirmUntilCreateCommit) + { + NProto::TStorageConfig storageConfig; + storageConfig.SetAsyncCreateHandleEnabled(true); + TTestEnv env({}, storageConfig); + + ui32 nodeIdx = env.AddDynamicNode(); + ui64 tabletId = env.BootIndexTablet(nodeIdx); + + TIndexTabletClient tablet(env.GetRuntime(), nodeIdx, tabletId); + tablet.InitSession("client", "session"); + + auto id = CreateNode(tablet, TCreateNodeArgs::File(RootNodeId, "test")); + + // Hold the original create tx commit. ConfirmCreateHandle may execute + // while the handle exists only in tablet memory, but it must not reply. + TAutoPtr putEvent; + env.GetRuntime().SetEventFilter( + [&](auto& runtime, auto& ev) + { + Y_UNUSED(runtime); + if (!putEvent && + ev->GetTypeRewrite() == TEvBlobStorage::EvPut) + { + putEvent = std::move(ev); + return true; + } + + return false; + }); + + auto request = + tablet.CreateCreateHandleRequest(id, TCreateHandleArgs::RDNLY); + request->Record.SetAllowAsyncCreateHandle(true); + request->Record.MutableHeaders()->SetRequestId(100500); + tablet.SendRequest(std::move(request)); + + env.GetRuntime().DispatchEvents(TDispatchOptions{ + .CustomFinalCondition = [&]() + { + return putEvent != nullptr; + }}); + + auto response = tablet.RecvCreateHandleResponse(); + UNIT_ASSERT_VALUES_EQUAL_C( + S_OK, + response->Record.GetError().GetCode(), + response->Record.GetError().GetMessage()); + UNIT_ASSERT(response->Record.GetHandleCreatedAsync()); + + const ui64 handle = response->Record.GetHandle(); + UNIT_ASSERT(handle); + + auto confirmRequest = tablet.CreateConfirmCreateHandleRequest( + id, + handle, + TCreateHandleArgs::RDNLY, + 100500); + tablet.SendRequest(std::move(confirmRequest)); + + // Confirm must not return success while the original async create tx is + // still uncommitted. Otherwise vhost could pop the durable queue entry + // before the handle is actually reloadable. + tablet.AssertConfirmCreateHandleNoResponse(); + + env.GetRuntime().SetEventFilter( + TTestActorRuntimeBase::DefaultFilterFunc); + env.GetRuntime().Send(putEvent.Release(), nodeIdx); + + auto confirmResponse = tablet.RecvConfirmCreateHandleResponse(); + UNIT_ASSERT_VALUES_EQUAL_C( + S_OK, + confirmResponse->Record.GetError().GetCode(), + confirmResponse->Record.GetError().GetMessage()); + + // Once confirm has replied, the preceding create commit is durable + // enough for the handle to be reloaded. + tablet.RebootTablet(); + tablet.RecoverSession(); + + tablet.DescribeData(handle, 0, 1_KB); + tablet.DestroyHandle(handle); + } + + Y_UNIT_TEST(ShouldLoadConfirmedCreateHandleAfterTabletRestart) + { + TTestEnv env; + + ui32 nodeIdx = env.AddDynamicNode(); + ui64 tabletId = env.BootIndexTablet(nodeIdx); + + TIndexTabletClient tablet(env.GetRuntime(), nodeIdx, tabletId); + tablet.InitSession("client", "session"); + + auto id = CreateNode(tablet, TCreateNodeArgs::File(RootNodeId, "test")); + const ui64 handle = 424244; + + tablet.ConfirmCreateHandle( + id, + handle, + TCreateHandleArgs::RDNLY, + 100500); + + // A confirmed handle must be present after restart, proving that the + // SessionHandles row was committed and reloaded. + tablet.RebootTablet(); + tablet.RecoverSession(); + + tablet.DescribeData(handle, 0, 1_KB); + tablet.DestroyHandle(handle); + } + + Y_UNIT_TEST(ShouldRejectConfirmCreateHandleOnCollision) + { + TTestEnv env; + + ui32 nodeIdx = env.AddDynamicNode(); + ui64 tabletId = env.BootIndexTablet(nodeIdx); + + TIndexTabletClient tablet(env.GetRuntime(), nodeIdx, tabletId); + tablet.InitSession("client", "session"); + + auto node1 = + CreateNode(tablet, TCreateNodeArgs::File(RootNodeId, "test1")); + auto node2 = + CreateNode(tablet, TCreateNodeArgs::File(RootNodeId, "test2")); + auto handle = CreateHandle(tablet, node1, {}, TCreateHandleArgs::RDNLY); + + auto response = tablet.AssertConfirmCreateHandleFailed( + node2, + handle, + TCreateHandleArgs::RDNLY, + 100500); + UNIT_ASSERT_VALUES_EQUAL_C( + E_INVALID_STATE, + response->Record.GetError().GetCode(), + response->Record.GetError().GetMessage()); + + tablet.DestroyHandle(handle); + } + + Y_UNIT_TEST(ShouldRejectConfirmCreateHandleForAnotherSessionHandle) + { + TTestEnv env; + + ui32 nodeIdx = env.AddDynamicNode(); + ui64 tabletId = env.BootIndexTablet(nodeIdx); + + TIndexTabletClient tablet1(env.GetRuntime(), nodeIdx, tabletId); + tablet1.InitSession("client1", "session1"); + + TIndexTabletClient tablet2(env.GetRuntime(), nodeIdx, tabletId); + tablet2.InitSession("client2", "session2"); + + auto node = + CreateNode(tablet1, TCreateNodeArgs::File(RootNodeId, "test")); + auto handle = + CreateHandle(tablet1, node, {}, TCreateHandleArgs::RDNLY); + + auto response = tablet2.AssertConfirmCreateHandleFailed( + node, + handle, + TCreateHandleArgs::RDNLY, + 100500); + UNIT_ASSERT_VALUES_EQUAL_C( + E_INVALID_STATE, + response->Record.GetError().GetCode(), + response->Record.GetError().GetMessage()); + + tablet1.DestroyHandle(handle); + } + + Y_UNIT_TEST(ShouldRejectConfirmCreateHandleForUnlinkedNode) + { + TTestEnv env; + + ui32 nodeIdx = env.AddDynamicNode(); + ui64 tabletId = env.BootIndexTablet(nodeIdx); + + TIndexTabletClient tablet(env.GetRuntime(), nodeIdx, tabletId); + tablet.InitSession("client", "session"); + + auto node = + CreateNode(tablet, TCreateNodeArgs::File(RootNodeId, "test")); + tablet.UnlinkNode(RootNodeId, "test", false); + tablet.AssertGetNodeAttrFailed(node); + + const ui64 handle = 424245; + auto response = tablet.AssertConfirmCreateHandleFailed( + node, + handle, + TCreateHandleArgs::RDNLY, + 100500); + UNIT_ASSERT_VALUES_EQUAL_C( + E_FS_NOENT, + response->Record.GetError().GetCode(), + response->Record.GetError().GetMessage()); + } + Y_UNIT_TEST(ShouldSetGuestKeepCacheProperlyForOffloadedNodes) { NProto::TStorageConfig storageConfig; diff --git a/cloud/filestore/libs/storage/testlib/tablet_client.h b/cloud/filestore/libs/storage/testlib/tablet_client.h index cfab5835de3..bd34bdafc4d 100644 --- a/cloud/filestore/libs/storage/testlib/tablet_client.h +++ b/cloud/filestore/libs/storage/testlib/tablet_client.h @@ -861,6 +861,21 @@ class TIndexTabletClient return request; } + auto CreateConfirmCreateHandleRequest( + ui64 node, + ui64 handle, + ui32 flags, + ui64 requestId = 0) + { + auto request = + CreateSessionRequest(); + request->Record.SetNodeId(node); + request->Record.SetHandle(handle); + request->Record.SetFlags(flags); + request->Record.SetRequestId(requestId); + return request; + } + auto CreateDestroyHandleRequest(ui64 handle) { auto request = CreateSessionRequest();