[Filestore] issue-2189: support O_DIRECT in local filestore (#2188)

cloud/filestore/config/filesystem.proto

@@ -42,4 +42,10 @@ message TFileSystemConfig
 
     // Period of processing create/destroy handle requests.
     optional uint32 AsyncHandleOperationPeriod = 12;
+
+    // Enable O_DIRECT when working with files
+    optional bool DirectIoEnabled = 13;
+
+    // Aligment needed for buffers when using direct io
+    optional uint32 DirectIoAlign = 14;
 }

cloud/filestore/config/server.proto

@@ -83,6 +83,12 @@ message TLocalServiceConfig
     // Maximum number of file handles which can be opened by single session
     // for single local file system
     optional uint32 MaxHandlePerSessionCount = 8;
+
+    // Enable O_DIRECT when working with files
+    optional bool DirectIoEnabled = 9;
+
+    // Aligment needed for buffers when using direct io
+    optional uint32 DirectIoAlign = 10;
 }
 
 ////////////////////////////////////////////////////////////////////////////////

cloud/filestore/libs/service/filestore.cpp

@@ -33,6 +33,7 @@ constexpr std::array SUPPORTED_HANDLE_FLAGS = {
     TFlag2Proto{O_NOFOLLOW,  TCreateHandleRequest::E_NOFOLLOW},
     TFlag2Proto{O_NONBLOCK,  TCreateHandleRequest::E_NONBLOCK},
     TFlag2Proto{O_PATH,      TCreateHandleRequest::E_PATH},
+    TFlag2Proto{O_DIRECT,    TCreateHandleRequest::E_DIRECT},
 };
 
 constexpr std::array SUPPORTED_RENAME_FLAGS = {

cloud/filestore/libs/service_local/config.cpp

@@ -2,6 +2,8 @@
 
 #include <library/cpp/monlib/service/pages/templates.h>
 
+#include <util/generic/size_literals.h>
+
 namespace NCloud::NFileStore {
 
 namespace {
@@ -17,6 +19,8 @@ namespace {
     xxx(StatePath,                   TString,       "./"                      )\
     xxx(MaxNodeCount,                ui32,          1000000                   )\
     xxx(MaxHandlePerSessionCount,    ui32,          10000                     )\
+    xxx(DirectIoEnabled,             bool,          false                     )\
+    xxx(DirectIoAlign,               ui32,          4_KB                      )\
 // FILESTORE_SERVICE_CONFIG
 
 #define FILESTORE_SERVICE_DECLARE_CONFIG(name, type, value)                    \

cloud/filestore/libs/service_local/config.h

@@ -29,6 +29,8 @@ class TLocalFileStoreConfig
     TString GetStatePath() const;
     ui32 GetMaxNodeCount() const;
     ui32 GetMaxHandlePerSessionCount() const;
+    bool GetDirectIoEnabled() const;
+    ui32 GetDirectIoAlign() const;
 
     void Dump(IOutputStream& out) const;
     void DumpHtml(IOutputStream& out) const;

cloud/filestore/libs/service_local/fs.cpp

@@ -24,6 +24,11 @@ TLocalFileSystem::TLocalFileSystem(
 {
     Log = Logging->CreateLog(Store.GetFileSystemId());
 
+    STORAGE_INFO(
+        "LocalFileSystemId=" << Store.GetFileSystemId() <<
+        ", DirectIoEnabled=" << Config->GetDirectIoEnabled() <<
+        ", DirectIoAlign=" << Config->GetDirectIoAlign());
+
     Index = std::make_shared<TLocalIndex>(
         Root,
         StatePath,

cloud/filestore/libs/service_local/fs_data.cpp

@@ -3,6 +3,8 @@
 #include "lowlevel.h"
 
 #include <cloud/filestore/libs/diagnostics/critical_events.h>
+
+#include <cloud/storage/core/libs/common/aligned_buffer.h>
 #include <cloud/storage/core/libs/common/file_io_service.h>
 
 #include <util/string/builder.h>
@@ -30,6 +32,9 @@ NProto::TCreateHandleResponse TLocalFileSystem::CreateHandle(
     }
 
     int flags = HandleFlagsToSystem(request.GetFlags());
+    if (!Config->GetDirectIoEnabled()) {
+        flags &= ~O_DIRECT;
+    }
     const int mode = request.GetMode()
         ? request.GetMode() : Config->GetDefaultPermissions();
 
@@ -97,18 +102,23 @@ TFuture<NProto::TReadDataResponse> TLocalFileSystem::ReadDataAsync(
             TErrorResponse(ErrorInvalidHandle(request.GetHandle())));
     }
 
-    auto b = TString::Uninitialized(request.GetLength());
-    NSan::Unpoison(b.data(), b.size());
+    auto align = Config->GetDirectIoEnabled() ? Config->GetDirectIoAlign() : 0;
+    auto b = std::make_shared<TAlignedBuffer>(request.GetLength(), align);
+    NSan::Unpoison(b->Begin(), b->Size());
 
-    TArrayRef<char> data(b.begin(), b.vend());
+    TArrayRef<char> data(b->Begin(), b->End());
     auto promise = NewPromise<NProto::TReadDataResponse>();
     FileIOService->AsyncRead(*handle, request.GetOffset(), data).Subscribe(
         [b = std::move(b), promise] (const TFuture<ui32>& f) mutable {
             NProto::TReadDataResponse response;
             try {
                 auto bytesRead = f.GetValue();
-                b.resize(bytesRead);
-                response.SetBuffer(std::move(b));
+                b->TrimSize(bytesRead);
+                response.SetBufferOffset(b->AlignedDataOffset());
+                response.SetBuffer(std::move(b->AccessBuffer()));
+            } catch (const TServiceError& e) {
+                *response.MutableError() = MakeError(MAKE_FILESTORE_ERROR(
+                    ErrnoToFileStoreError(STATUS_FROM_CODE(e.GetCode()))));
             } catch (...) {
                 *response.MutableError() =
                     MakeError(E_IO, CurrentExceptionMessage());
@@ -131,13 +141,17 @@ TFuture<NProto::TWriteDataResponse> TLocalFileSystem::WriteDataAsync(
     }
 
     auto b = std::move(*request.MutableBuffer());
-    TArrayRef<char> data(b.begin(), b.vend());
+    auto offset = request.GetBufferOffset();
+    TArrayRef<char> data(b.begin() + offset, b.vend());
     auto promise = NewPromise<NProto::TWriteDataResponse>();
     FileIOService->AsyncWrite(*handle, request.GetOffset(), data).Subscribe(
         [b = std::move(b), promise] (const TFuture<ui32>& f) mutable {
             NProto::TWriteDataResponse response;
             try {
                 f.GetValue();
+            } catch (const TServiceError& e) {
+                *response.MutableError() = MakeError(MAKE_FILESTORE_ERROR(
+                    ErrnoToFileStoreError(STATUS_FROM_CODE(e.GetCode()))));
             } catch (...) {
                 *response.MutableError() =
                     MakeError(E_IO, CurrentExceptionMessage());

cloud/filestore/libs/service_local/fs_session.cpp

@@ -75,6 +75,10 @@ NProto::TCreateSessionResponse TLocalFileSystem::CreateSession(
     NProto::TCreateSessionResponse response;
     session->GetInfo(*response.MutableSession(), sessionSeqNo);
 
+    auto* features = response.MutableFileStore()->MutableFeatures();
+    features->SetDirectIoEnabled(Config->GetDirectIoEnabled());
+    features->SetDirectIoAlign(Config->GetDirectIoAlign());
+
     return response;
 }
 

cloud/filestore/libs/service_local/service_ut.cpp

@@ -6,6 +6,7 @@
 #include <cloud/filestore/libs/service/filestore.h>
 
 #include <cloud/storage/core/libs/aio/service.h>
+#include <cloud/storage/core/libs/common/aligned_buffer.h>
 #include <cloud/storage/core/libs/common/error.h>
 #include <cloud/storage/core/libs/common/file_io_service.h>
 #include <cloud/storage/core/libs/common/scheduler.h>
@@ -229,6 +230,9 @@ struct TCreateHandleArgs
         = ProtoFlag(NProto::TCreateHandleRequest::E_CREATE)
         | ProtoFlag(NProto::TCreateHandleRequest::E_READ)
         | ProtoFlag(NProto::TCreateHandleRequest::E_WRITE);
+
+    static constexpr ui32 DIRECT
+        = ProtoFlag(NProto::TCreateHandleRequest::E_DIRECT);
 };
 
 ////////////////////////////////////////////////////////////////////////////////
@@ -289,12 +293,18 @@ struct TTestBootstrap
     TTestBootstrap(
             const TTempDirectoryPtr& cwd = std::make_shared<TTempDirectory>(),
             ui32 maxNodeCount = 1000,
-            ui32 maxHandlePerSessionCount = 100)
+            ui32 maxHandlePerSessionCount = 100,
+            bool directIoEnabled = false,
+            ui32 directIoAlign = 4096)
         : Cwd(cwd)
     {
         AIOService->Start();
         Store = CreateLocalFileStore(
-            CreateConfig(maxNodeCount, maxHandlePerSessionCount),
+            CreateConfig(
+                maxNodeCount,
+                maxHandlePerSessionCount,
+                directIoEnabled,
+                directIoAlign),
             Timer,
             Scheduler,
             Logging,
@@ -308,12 +318,18 @@ struct TTestBootstrap
             const TString& client = "client",
             const TString& session = {},
             ui32 maxNodeCount = 1000,
-            ui32 maxHandlePerSessionCount = 100)
+            ui32 maxHandlePerSessionCount = 100,
+            bool directIoEnabled = false,
+            ui32 directIoAlign = 4096)
         : Cwd(std::make_shared<TTempDirectory>())
     {
         AIOService->Start();
         Store = CreateLocalFileStore(
-            CreateConfig(maxNodeCount, maxHandlePerSessionCount),
+            CreateConfig(
+                maxNodeCount,
+                maxHandlePerSessionCount,
+                directIoEnabled,
+                directIoAlign),
             Timer,
             Scheduler,
             Logging,
@@ -373,13 +389,17 @@ struct TTestBootstrap
 
     TLocalFileStoreConfigPtr CreateConfig(
         ui32 maxNodeCount,
-        ui32 maxHandlePerSessionCount)
+        ui32 maxHandlePerSessionCount,
+        bool directIoEnabled,
+        ui32 directIoAlign)
     {
         NProto::TLocalServiceConfig config;
         config.SetRootPath(Cwd->GetName());
         config.SetStatePath(Cwd->GetName());
         config.SetMaxNodeCount(maxNodeCount);
         config.SetMaxHandlePerSessionCount(maxHandlePerSessionCount);
+        config.SetDirectIoEnabled(directIoEnabled);
+        config.SetDirectIoAlign(directIoAlign);
 
         return std::make_shared<TLocalFileStoreConfig>(config);
     }
@@ -746,6 +766,64 @@ struct TTestBootstrap
     FILESTORE_SERVICE(FILESTORE_DECLARE_METHOD)
 
 #undef FILESTORE_DECLARE_METHOD
+
+    NProto::TWriteDataResponse WriteDataAligned(
+        ui64 handle,
+        ui64 offset,
+        const TString& buffer,
+        ui32 align)
+    {
+        auto request = CreateRequest<NProto::TWriteDataRequest>();
+        request->SetHandle(handle);
+        request->SetOffset(offset);
+
+        TAlignedBuffer alignedBuffer(buffer.size(), align);
+        memcpy(
+            (void*)(alignedBuffer.Begin()),
+            (void*)buffer.data(),
+            buffer.size());
+        request->SetBufferOffset(alignedBuffer.AlignedDataOffset());
+        request->SetBuffer(std::move(alignedBuffer.AccessBuffer()));
+
+        auto dbg = request->ShortDebugString();
+        auto response =
+            Store->WriteData(Ctx, std::move(request)).GetValueSync();
+
+        UNIT_ASSERT_C(
+            SUCCEEDED(response.GetError().GetCode()),
+            DumpMessage(response.GetError()) + "@" + dbg);
+
+        return response;
+    }
+
+    NProto::TWriteDataResponse AssertWriteDataAlignedFailed(
+        ui64 handle,
+        ui64 offset,
+        const TString& buffer,
+        ui32 align)
+    {
+        auto request = CreateRequest<NProto::TWriteDataRequest>();
+        request->SetHandle(handle);
+        request->SetOffset(offset);
+
+        TAlignedBuffer alignedBuffer(buffer.size(), align);
+        memcpy(
+            (void*)(alignedBuffer.Begin()),
+            (void*)buffer.data(),
+            buffer.size());
+        request->SetBufferOffset(alignedBuffer.AlignedDataOffset());
+        request->SetBuffer(std::move(alignedBuffer.AccessBuffer()));
+
+        auto dbg = request->ShortDebugString();
+        auto response =
+            Store->WriteData(Ctx, std::move(request)).GetValueSync();
+
+        UNIT_ASSERT_C(
+            FAILED(response.GetError().GetCode()),
+            "WriteDataAligned has not failed as expected " + dbg);
+
+        return response;
+    }
 };
 
 ////////////////////////////////////////////////////////////////////////////////
@@ -1939,6 +2017,76 @@ Y_UNIT_TEST_SUITE(LocalFileStore)
             bootstrap.FsyncDir(dirNodeId, dataSync);
         }
     }
+
+    void CheckReadAndWriteDataWithDirectIo(ui32 directIoAlign)
+    {
+        TTestBootstrap bootstrap(
+            "fs",
+            "client",
+            {},
+            1000,
+            1000,
+            true /* direct io enabled */,
+            directIoAlign);
+
+        ui64 handle =
+            bootstrap
+                .CreateHandle(
+                    RootNodeId,
+                    "file",
+                    TCreateHandleArgs::CREATE | TCreateHandleArgs::DIRECT)
+                .GetHandle();
+        auto readRsp = bootstrap.ReadData(handle, 0, 100);
+        auto data = readRsp.GetBuffer().substr(readRsp.GetBufferOffset());
+        UNIT_ASSERT_VALUES_EQUAL(data.size(), 0);
+
+        data = "aaaabbbbcccccdddddeeee";
+        const auto response = bootstrap.AssertWriteDataAlignedFailed(
+            handle,
+            0,
+            data,
+            directIoAlign);
+        const auto& error = response.GetError();
+        UNIT_ASSERT_VALUES_EQUAL(
+            static_cast<ui32>(NProto::E_FS_INVAL),
+            STATUS_FROM_CODE(error.GetCode()));
+
+        data.append(directIoAlign-data.size(), 'x');
+        data.append(directIoAlign, 'y');
+        bootstrap.WriteDataAligned(handle, 0, data, directIoAlign);
+
+        auto readDataWithOffset =
+            [&bootstrap](ui64 handle, ui64 offset, ui32 len)
+        {
+            auto rsp = bootstrap.ReadData(handle, offset, len);
+            return rsp.GetBuffer().substr(rsp.GetBufferOffset());
+        };
+
+        // read [0, 2*align]
+        auto buffer = readDataWithOffset(handle, 0, data.size());
+        UNIT_ASSERT_VALUES_EQUAL(buffer, data);
+
+        // read [0, align]
+        buffer = readDataWithOffset(handle, 0, directIoAlign);
+        UNIT_ASSERT_VALUES_EQUAL(buffer, data.substr(0, directIoAlign));
+
+        // read [align, align]
+        buffer = readDataWithOffset(handle, directIoAlign, directIoAlign);
+        UNIT_ASSERT_VALUES_EQUAL(
+            buffer,
+            data.substr(directIoAlign, directIoAlign));
+    }
+
+    Y_UNIT_TEST(ShouldReadAndWriteDataWithDirectIoAlignedTo512)
+    {
+        CheckReadAndWriteDataWithDirectIo(512);
+    }
+
+    Y_UNIT_TEST(ShouldReadAndWriteDataWithDirectIoAlignedTo4096)
+    {
+        CheckReadAndWriteDataWithDirectIo(4096);
+    }
+
 };
 
 }   // namespace NCloud::NFileStore

cloud/filestore/libs/vfs_fuse/config.cpp

@@ -29,6 +29,9 @@ namespace {
                                                                                \
     xxx(AsyncDestroyHandleEnabled,  bool,       false                         )\
     xxx(AsyncHandleOperationPeriod, TDuration,  TDuration::MilliSeconds(50)   )\
+                                                                               \
+    xxx(DirectIoEnabled,            bool,       false                         )\
+    xxx(DirectIoAlign,              ui32,       4_KB                          )\
 // FILESTORE_FUSE_CONFIG
 
 #define FILESTORE_FILESYSTEM_DECLARE_CONFIG(name, type, value)                 \

cloud/filestore/libs/vfs_fuse/config.h

@@ -39,6 +39,9 @@ struct TFileSystemConfig
     bool GetAsyncDestroyHandleEnabled() const;
     TDuration GetAsyncHandleOperationPeriod() const;
 
+    bool GetDirectIoEnabled() const;
+    ui32 GetDirectIoAlign() const;
+
     void Dump(IOutputStream& out) const;
     void DumpHtml(IOutputStream& out) const;
 };