Commit 6ad69213 authored by Ramon Nou's avatar Ramon Nou
Browse files

Merge branch 'rnou/directio' into 'master' 

Resolve "Syscall + libfabric bug on buffer reuse"

Adds also DIRECT_IO on the server side to increase I/O performance.
DIRECT_IO is used only on full chunks. There is a fallback to non direct-io on non-aligned writes.
Closes #390

Closes #390

See merge request !312
parents 0185d5bc 16b315c6

CHANGELOG.md

+2 −0
Original line number Diff line number Diff line
@@ -56,6 +56,7 @@ to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). 
  - Moved some CMAKE options to config.hpp and env variables ([!285](https://storage.bsc.es/gitlab/hpc/gekkofs/-/merge_requests/285))

    - LIBGKFS/ GKFS _SYMLINK_SUPPORT, _RENAME_SUPPORT and _CREATE_CHECK_PARENTS.

    - Now all the performance options are in config.hpp and env variables.

  - Added DIRECT_IO on server side to increase performance ([!312](https://storage.bsc.es/gitlab/hpc/gekkofs/-/merge_requests/312))



### Fixed

  - SYS_lstat does not exists on some architectures, change to newfstatat ([!269](https://storage.bsc.es/gitlab/hpc/gekkofs/-/merge_requests/269))
@@ -65,6 +66,7 @@ to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). 
  - Fix cuda in syscall ([!292](https://storage.bsc.es/gitlab/hpc/gekkofs/-/merge_requests/292))

    - mmap and dangling fd issues

  - Fix remove chunk bug ([!294](https://storage.bsc.es/gitlab/hpc/gekkofs/-/merge_requests/294))

  - Fix decompress_and_parse_entries_standard() Unexpected end of buffer while parsing name bug ([!312](https://storage.bsc.es/gitlab/hpc/gekkofs/-/merge_requests/312))

    



## [0.9.5] - 2025-08

src/daemon/backend/data/chunk_storage.cpp

+85 −11
Original line number Diff line number Diff line
@@ -155,6 +155,15 @@ ChunkStorage::destroy_chunk_space(const string& file_path) const { 
 * for pwrite behavior.

 * @endinternal

 */

static bool

is_direct_io_compatible(const char* buf, size_t size, off64_t offset) {

    // O_DIRECT requires alignment to disk_min_io (usually 512 bytes)

    bool ptr_aligned = (reinterpret_cast<uintptr_t>(buf) % 512 == 0);

    bool offset_aligned = (offset % 512 == 0);

    bool size_aligned = (size % 512 == 0);

    return ptr_aligned && offset_aligned && size_aligned;

}



ssize_t

ChunkStorage::write_chunk(const string& file_path,

                          gkfs::rpc::chnk_id_t chunk_id, const char* buf,
@@ -170,8 +179,17 @@ ChunkStorage::write_chunk(const string& file_path, 
        chunk_path = absolute(get_chunk_path(file_path, chunk_id));

    }



    FileHandle fh(open(chunk_path.c_str(), O_WRONLY | O_CREAT, 0640),

                  chunk_path);

    // Check if we can use O_DIRECT (requires 512-byte aligned buf, offset,

    // size)

    bool can_direct = is_direct_io_compatible(buf, size, offset);



    int open_flags_write = O_WRONLY | O_CREAT;

    if(can_direct) {

#ifdef O_DIRECT

        open_flags_write |= O_DIRECT;

#endif

    }

    FileHandle fh(open(chunk_path.c_str(), open_flags_write, 0640), chunk_path);

    if(!fh.valid()) {

        auto err_str = fmt::format(

                "{}() Failed to open chunk file for write. File: '{}', Error: '{}'",
@@ -179,14 +197,34 @@ ChunkStorage::write_chunk(const string& file_path, 
        throw ChunkStorageException(errno, err_str);

    }



    // Use aligned buffer if O_DIRECT not compatible

    static thread_local char aligned_buf[65536]; // max common write size

    char* tmp_aligned = nullptr;

    const char* write_buf = buf;

    size_t write_size = size;

    if(!can_direct) {

        if(size <= sizeof(aligned_buf)) {

            std::memcpy(aligned_buf, buf, size);

            write_buf = aligned_buf;

        } else {

            // For large writes, allocate on heap with alignment

            tmp_aligned = static_cast<char*>(aligned_alloc(512, size));

            if(tmp_aligned) {

                std::memcpy(tmp_aligned, buf, size);

                write_buf = tmp_aligned;

                write_size = size;

            }

            // Fall back to normal write without O_DIRECT

        }

    }



    size_t wrote_total{};

    do {

        ssize_t wrote = pwrite(fh.native(), buf + wrote_total,

                               size - wrote_total, offset + wrote_total);

        ssize_t wrote = pwrite(fh.native(), write_buf + wrote_total,

                               write_size - wrote_total, offset + wrote_total);



        if(wrote < 0) {

            // retry if a signal or anything else has interrupted the read

            // system call

            // retry if a signal or anything else has interrupted the write

            if(errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK)

                continue;

            auto err_str = fmt::format(
@@ -195,9 +233,10 @@ ChunkStorage::write_chunk(const string& file_path, 
            throw ChunkStorageException(errno, err_str);

        }

        wrote_total += wrote;

    } while(wrote_total != size);

    } while(wrote_total != write_size);



    // file is closed via the file handle's destructor.

    free(tmp_aligned); // Clean up tmp if allocated

    return wrote_total;

}
@@ -219,17 +258,46 @@ ChunkStorage::read_chunk(const string& file_path, gkfs::rpc::chnk_id_t chunk_id, 
        chunk_path = absolute(get_chunk_path(file_path, chunk_id));

    }



    FileHandle fh(open(chunk_path.c_str(), O_RDONLY), chunk_path);

    // Check if we can use O_DIRECT (requires 512-byte aligned buf, offset,

    // size)

    bool can_direct = is_direct_io_compatible(buf, size, offset);



    int open_flags_read = O_RDONLY;

    if(can_direct) {

#ifdef O_DIRECT

        open_flags_read |= O_DIRECT;

#endif

    }

    FileHandle fh(open(chunk_path.c_str(), open_flags_read), chunk_path);

    if(!fh.valid()) {

        auto err_str = fmt::format(

                "{}() Failed to open chunk file for read. File: '{}', Error: '{}'",

                __func__, chunk_path, ::strerror(errno));

        throw ChunkStorageException(errno, err_str);

    }



    // Use aligned buffer if O_DIRECT not compatible

    static thread_local char aligned_buf[65536];

    char* tmp_aligned = nullptr;

    char* read_buf = buf;

    size_t read_size = size;

    if(!can_direct) {

        if(size <= sizeof(aligned_buf)) {

            read_buf = aligned_buf;

            read_size = size;

        } else {

            tmp_aligned = static_cast<char*>(aligned_alloc(512, size));

            if(tmp_aligned) {

                read_buf = tmp_aligned;

                read_size = size;

            }

        }

    }



    size_t read_total = 0;

    do {

        ssize_t read = pread64(fh.native(), buf + read_total, size - read_total,

                               offset + read_total);

        ssize_t read = pread64(fh.native(), read_buf + read_total,

                               read_size - read_total, offset + read_total);

        if(read == 0) {

            /*

             * A value of zero indicates end-of-file (except if the value of the
@@ -260,9 +328,15 @@ ChunkStorage::read_chunk(const string& file_path, gkfs::rpc::chnk_id_t chunk_id, 
#endif

        assert(read > 0);

        read_total += read;

    } while(read_total != size);

    } while(read_total != read_size);



    // Copy back to user buffer if we used a temp buffer

    if(read_buf != buf && !can_direct) {

        std::memcpy(buf, read_buf, read_total);

    }



    // file is closed via the file handle's destructor.

    free(tmp_aligned);

    return read_total;

}

src/daemon/handler/srv_metadata.cpp

+22 −7
Original line number Diff line number Diff line
@@ -553,15 +553,29 @@ get_dirents_helper(const std::shared_ptr<tl::engine>& engine, 
    // Calculate total output size

    size_t uncompressed_size = 0;

    size_t entries_serialized = 0;

    std::vector<char> uncompressed_data;

    auto append_bytes = [&uncompressed_data](const void* src, size_t len) {

        const auto old_size = uncompressed_data.size();

        uncompressed_data.resize(old_size + len);

        std::memcpy(uncompressed_data.data() + old_size, src, len);

    };



    // Use static thread_local buffers to avoid libfabric MR cache issues.

    // Local std::vector<char> buffers get their heap memory registered as

    // Memory Regions in libfabric. When the vector is destroyed and a new one

    // is created, the heap allocator may reuse the same address but the stale

    // MR cache entry causes RMA operations to read/write wrong memory.

    // Using thread_local buffers ensures stable addresses across calls.

    static thread_local std::vector<char> uncompressed_data;

    static thread_local std::vector<char> compressed_data;

    uncompressed_data.clear();

    compressed_data.clear();

    if(client_bulk_size > 0)

        uncompressed_data.reserve(client_bulk_size); // Hint for reservation



    // Lambda to append bytes to uncompressed_data (captured by pointer to avoid

    // thread_local capture warning)

    auto uncompressed_data_ptr = &uncompressed_data;

    auto append_bytes = [uncompressed_data_ptr](const void* src, size_t len) {

        const auto old_size = uncompressed_data_ptr->size();

        uncompressed_data_ptr->resize(old_size + len);

        std::memcpy(uncompressed_data_ptr->data() + old_size, src, len);

    };



    // Used for extended Dirents

    // TODO: This should be refactored to use a more generic approach

    if constexpr(std::is_same_v<EntryType,
@@ -649,7 +663,6 @@ get_dirents_helper(const std::shared_ptr<tl::engine>& engine, 
    }

    uncompressed_size = uncompressed_data.size();



    std::vector<char> compressed_data;

    void* segment_ptr = nullptr;

    size_t transfer_size = 0;
@@ -657,6 +670,8 @@ get_dirents_helper(const std::shared_ptr<tl::engine>& engine, 
    if(gkfs::config::rpc::use_dirents_compression) {

        const size_t compressed_bound = ZSTD_compressBound(uncompressed_size);

        compressed_data.resize(compressed_bound);

        // Ensure the buffer is zeroed to prevent stale data issues

        std::memset(compressed_data.data(), 0, compressed_bound);



        const size_t compressed_size =

                ZSTD_compress(compressed_data.data(), compressed_bound,