From 16b315c6fe17a543fe7685b3a010f87040261d5a Mon Sep 17 00:00:00 2001 From: Ramon Nou Date: Thu, 13 Aug 2026 20:42:35 +0200 Subject: [PATCH] DIRECT_IO and fix libfabric error --- CHANGELOG.md | 2 + src/daemon/backend/data/chunk_storage.cpp | 96 ++++++++++++++++++++--- src/daemon/handler/srv_metadata.cpp | 29 +++++-- 3 files changed, 109 insertions(+), 18 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 97c74618b..77c936fb4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -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 diff --git a/src/daemon/backend/data/chunk_storage.cpp b/src/daemon/backend/data/chunk_storage.cpp index 4285e6407..061000654 100644 --- a/src/daemon/backend/data/chunk_storage.cpp +++ b/src/daemon/backend/data/chunk_storage.cpp @@ -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(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(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(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; } diff --git a/src/daemon/handler/srv_metadata.cpp b/src/daemon/handler/srv_metadata.cpp index 98e05c2a2..e7522cc27 100644 --- a/src/daemon/handler/srv_metadata.cpp +++ b/src/daemon/handler/srv_metadata.cpp @@ -553,15 +553,29 @@ get_dirents_helper(const std::shared_ptr& engine, // Calculate total output size size_t uncompressed_size = 0; size_t entries_serialized = 0; - std::vector 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 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 uncompressed_data; + static thread_local std::vector 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& engine, } uncompressed_size = uncompressed_data.size(); - std::vector compressed_data; void* segment_ptr = nullptr; size_t transfer_size = 0; @@ -657,6 +670,8 @@ get_dirents_helper(const std::shared_ptr& 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, -- GitLab