mooncake-track
/
Mooncake
0
0
Fork 12
Code Issues Pull Requests 13 Actions Packages Projects Releases Wiki Activity

[Store]Optimize uring file support for SSD offloading (#1562) 

Co-authored-by: zhuxinjie-nz <240190801+zhuxinjie-nz@users.noreply.github.com>
This commit is contained in:
zhangzuo21 2026-03-12 12:34:44 +08:00 committed by GitHub
parent e43dd7dec1
commit e7f6bad7fa
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
13 changed files with 860 additions and 842 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

74
mooncake-store/include/file_interface.h
View File

@ -174,6 +174,15 @@ class PosixFile : public StorageFile {
};
#ifdef USE_URING
/**
* @class UringFile
* @brief StorageFile backed by a process-wide shared io_uring ring.
*
* All UringFile instances share a single SharedUringRing singleton, so
* construction and destruction only register/unregister an fd slot — no
* per-file io_uring_queue_init / io_uring_queue_exit (no mmap/munmap,
* no TLB shootdown).
*/
class UringFile : public StorageFile {
public:
UringFile(const std::string &filename, int fd, unsigned queue_depth = 32,
@ -198,57 +207,42 @@ class UringFile : public StorageFile {
size_t length,
off_t offset = 0);
// Flush data to stable storage via
// io_uring_prep_fsync(IORING_FSYNC_DATASYNC). Must be called after write
// and before writing dependent metadata files.
// Batch read: submit up to 32 independent reads at once (each at its own
// offset) before waiting for completions, giving NVMe queue depth > 1.
// Use in BatchLoadBucket instead of per-key read_aligned() loops.
struct ReadDesc {
void *buf;
size_t len;
off_t off;
};
tl::expected<size_t, ErrorCode> batch_read(const ReadDesc *descs, int cnt);
// Flush data to stable storage via IORING_FSYNC_DATASYNC.
// Must be called after write_aligned and before writing dependent metadata.
tl::expected<void, ErrorCode> datasync();
// Buffer registration interface for high-performance I/O
// Register a single buffer with io_uring to avoid get_user_pages() overhead
// Returns true on success, false on failure
// Buffer registration — delegates to the shared ring (process-wide).
// Static variant: no file instance needed. Must be called once from a
// single thread before I/O threads begin. Other threads lazily pick up
// the registration on their first I/O call via ensure_buf_registered().
static bool register_global_buffer(void *buffer, size_t length);
static void unregister_global_buffer();
bool register_buffer(void *buffer, size_t length);
// Unregister previously registered buffer
void unregister_buffer();
// Check if a buffer is currently registered
bool is_buffer_registered() const { return buffer_registered_; }
bool is_buffer_registered() const;
private:
struct io_uring ring_;
bool ring_initialized_;
bool files_registered_;
bool buffer_registered_;
unsigned queue_depth_;
bool use_direct_io_;
static constexpr size_t ALIGNMENT_ =
4096; // O_DIRECT alignment requirement
static constexpr size_t ALIGNMENT_ = 4096;
// Registered buffer info
void *registered_buffer_;
size_t registered_buffer_size_;
struct iovec registered_iovec_;
/// Submit all pending SQEs and wait for exactly @p n completions.
/// Returns the total bytes transferred, or an error.
tl::expected<size_t, ErrorCode> submit_and_wait_n(int n);
/// Calculate optimal chunk size for parallel I/O based on:
/// - total_len: remaining bytes to transfer
/// - available_depth: number of queue slots available
/// - min_chunk_size: minimum chunk size (must be power of 2)
/// Returns a power-of-2 chunk size that maximizes queue utilization.
size_t calculate_chunk_size(size_t total_len, unsigned available_depth,
size_t min_chunk_size) const;
/// Allocate aligned buffer for O_DIRECT
/// Allocate / free an O_DIRECT aligned bounce buffer.
void *alloc_aligned_buffer(size_t size) const;
/// Free aligned buffer
void free_aligned_buffer(void *ptr) const;
/// Mutex to serialize concurrent access to ring_
mutable Mutex ring_mutex_;
/// Return true if @p buf falls entirely within the shared registered
/// buffer.
bool in_registered_buffer(const void *buf, size_t len) const;
};
#endif // USE_URING

29
mooncake-store/include/file_storage.h
View File

@ -14,30 +14,48 @@ class FileStorage {
tl::expected<void, ErrorCode> Init();
/**
* @brief Result of BatchGet operation containing batch_id and buffer
* pointers.
*/
struct BatchGetResult {
uint64_t batch_id;
std::vector<uint64_t> pointers;
};
/**
* @brief Reads multiple key-value (KV) entries from local storage and
* forwards them to a remote node.
* @param keys List of keys to read from the local KV store
* @param sizes Expected size in bytes for each value
* @return tl::expected<std::vector<uint64_t>, ErrorCode> indicating
* operation status.
* @return tl::expected<BatchGetResult, ErrorCode> containing batch_id and
* buffer pointers.
*/
tl::expected<std::vector<uint64_t>, ErrorCode> BatchGet(
tl::expected<BatchGetResult, ErrorCode> BatchGet(
const std::vector<std::string>& keys,
const std::vector<int64_t>& sizes);
FileStorageConfig config_;
/**
* @brief Releases buffer associated with a specific batch_id.
* Called by remote client after transfer completion.
* @param batch_id The unique identifier of the batch to release
* @return true if batch was found and released, false otherwise
*/
bool ReleaseBuffer(uint64_t batch_id);
private:
friend class FileStorageTest;
struct AllocatedBatch {
uint64_t batch_id;
std::vector<BufferHandle> handles;
std::unordered_map<std::string, Slice> slices;
std::chrono::steady_clock::time_point lease_timeout;
std::vector<uint64_t> pointers;
uint64_t total_size;
AllocatedBatch() = default;
AllocatedBatch() : batch_id(0), total_size(0) {}
AllocatedBatch(AllocatedBatch&&) = default;
AllocatedBatch& operator=(AllocatedBatch&&) = default;
@ -86,8 +104,9 @@ class FileStorage {
std::shared_ptr<StorageBackendInterface> storage_backend_;
std::shared_ptr<ClientBufferAllocator> client_buffer_allocator_;
mutable Mutex client_buffer_mutex_;
std::vector<std::shared_ptr<AllocatedBatch>> GUARDED_BY(
std::unordered_map<uint64_t, std::shared_ptr<AllocatedBatch>> GUARDED_BY(
client_buffer_mutex_) client_buffer_allocated_batches_;
std::atomic<uint64_t> next_batch_id_{1};
mutable Mutex offloading_mutex_;
bool GUARDED_BY(offloading_mutex_) enable_offloading_;

9
mooncake-store/include/master_service.h
View File

@ -712,6 +712,11 @@ class MasterService {
std::vector<ReplicaID> replica_ids;
};
struct OffloadingTask {
ReplicaID source_id;
std::chrono::system_clock::time_point start_time;
};
static constexpr size_t kNumShards = 1024; // Number of metadata shards
// Sharded metadata maps and their mutexes
@ -722,6 +727,8 @@ class MasterService {
std::unordered_set<std::string> processing_keys GUARDED_BY(mutex);
std::unordered_map<std::string, const ReplicationTask> replication_tasks
GUARDED_BY(mutex);
std::unordered_map<std::string, const OffloadingTask> offloading_tasks
GUARDED_BY(mutex);
};
std::array<MetadataShard, kNumShards> metadata_shards_;
@ -783,7 +790,7 @@ class MasterService {
void EvictionThreadFunc();
tl::expected<void, ErrorCode> PushOffloadingQueue(const std::string& key,
const Replica& replica);
Replica& replica);
// Lease related members
const uint64_t default_kv_lease_ttl_; // in milliseconds

10
mooncake-store/include/pyclient.h
View File

@ -73,6 +73,16 @@ class ClientRequester {
const std::vector<std::string> &keys,
const std::vector<int64_t> sizes);
/**
* @brief Notifies remote FileStorage to release buffer after transfer
* completion. This is a fire-and-forget call - errors are logged but not
* propagated.
* @param client_addr Network address of the remote FileStorage service.
* @param batch_id The batch_id returned from batch_get_offload_object.
*/
void release_offload_buffer(const std::string &client_addr,
uint64_t batch_id);
private:
/**
* @brief A batch of allocated memory buffers, tracking both handles and

8
mooncake-store/include/real_client.h
View File

@ -485,6 +485,14 @@ class RealClient : public PyClient {
batch_get_offload_object(const std::vector<std::string> &keys,
const std::vector<int64_t> &sizes);
/**
* @brief Releases buffer associated with a specific batch_id.
* Called by remote client after transfer completion.
* @param batch_id The unique identifier of the batch to release
* @return true if batch was found and released, false otherwise
*/
bool release_offload_buffer(uint64_t batch_id);
/**
* @brief Retrieves multiple stored objects from a remote service.
* @param target_rpc_service_addr Address of the remote RPC service (e.g.,

13
mooncake-store/include/rpc_types.h
View File

@ -144,19 +144,22 @@ struct TaskCompleteRequest {
YLT_REFL(TaskCompleteRequest, id, status, message);
struct BatchGetOffloadObjectResponse {
uint64_t batch_id;
std::vector<uint64_t> pointers;
std::string transfer_engine_addr;
uint64_t gc_ttl_ms;
BatchGetOffloadObjectResponse() = default;
BatchGetOffloadObjectResponse(std::vector<uint64_t>&& pointers_param,
BatchGetOffloadObjectResponse() : batch_id(0), gc_ttl_ms(0) {}
BatchGetOffloadObjectResponse(uint64_t batch_id_param,
std::vector<uint64_t>&& pointers_param,
std::string transfer_engine_addr_param,
uint64_t gc_ttl_ms_param)
: pointers(std::move(pointers_param)),
: batch_id(batch_id_param),
pointers(std::move(pointers_param)),
transfer_engine_addr(std::move(transfer_engine_addr_param)),
gc_ttl_ms(gc_ttl_ms_param) {}
};
YLT_REFL(BatchGetOffloadObjectResponse, pointers, transfer_engine_addr,
gc_ttl_ms);
YLT_REFL(BatchGetOffloadObjectResponse, batch_id, pointers,
transfer_engine_addr, gc_ttl_ms);
} // namespace mooncake

2
mooncake-store/include/storage_backend.h
View File

@ -903,7 +903,7 @@ class BucketStorageBackend : public StorageBackendInterface {
// Aligned buffer for O_DIRECT I/O operations
// We use a fixed-size buffer to avoid frequent allocations
static constexpr size_t kAlignedBufferSize = 16 * 1024 * 1024; // 16MB
static constexpr size_t kAlignedBufferSize = 32 * 1024 * 1024; // 16MB
std::unique_ptr<void, void (*)(void*)> aligned_io_buffer_{nullptr,
[](void*) {}};
/**

96
mooncake-store/src/file_storage.cpp
View File

@ -163,39 +163,22 @@ FileStorage::FileStorage(const FileStorageConfig& config,
storage_backend_ = create_storage_backend_result.value();
// Register buffer with UringFile if using BucketStorageBackend
// Register the client buffer with the process-wide io_uring fixed-buffer
// mechanism. This must happen before any I/O threads start so that they
// can lazily pick up the registration on their first I/O call.
#ifdef USE_URING
if (config.storage_backend_type == StorageBackendType::kBucket) {
auto bucket_backend =
std::dynamic_pointer_cast<BucketStorageBackend>(storage_backend_);
if (bucket_backend) {
auto file_result = bucket_backend->GetFileInstance();
if (file_result) {
auto file = file_result.value();
auto uring_file = std::dynamic_pointer_cast<UringFile>(file);
if (uring_file) {
auto aligned_allocator =
std::static_pointer_cast<AlignedClientBufferAllocator>(
client_buffer_allocator_);
if (aligned_allocator) {
void* base_ptr = aligned_allocator->get_base_pointer();
size_t size = aligned_allocator->get_total_size();
if (uring_file->register_buffer(base_ptr, size)) {
LOG(INFO)
<< "Successfully registered buffer with "
"UringFile: "
<< "base=" << base_ptr << ", size=" << size;
} else {
LOG(WARNING)
<< "Failed to register buffer with UringFile";
}
}
}
if (config.use_uring) {
auto aligned_allocator =
std::static_pointer_cast<AlignedClientBufferAllocator>(
client_buffer_allocator_);
if (aligned_allocator) {
void* base_ptr = aligned_allocator->get_base_pointer();
size_t size = aligned_allocator->get_total_size();
if (UringFile::register_global_buffer(base_ptr, size)) {
LOG(INFO) << "Successfully registered buffer with UringFile: "
<< "base=" << base_ptr << ", size=" << size;
} else {
LOG(WARNING)
<< "Failed to get file instance for buffer registration: "
<< file_result.error();
LOG(WARNING) << "Failed to register buffer with UringFile";
}
}
}
@ -284,7 +267,7 @@ tl::expected<void, ErrorCode> FileStorage::Init() {
return {};
}
tl::expected<std::vector<uint64_t>, ErrorCode> FileStorage::BatchGet(
tl::expected<FileStorage::BatchGetResult, ErrorCode> FileStorage::BatchGet(
const std::vector<std::string>& keys, const std::vector<int64_t>& sizes) {
auto start_time = std::chrono::steady_clock::now();
auto allocate_res = AllocateBatch(keys, sizes);
@ -310,15 +293,19 @@ tl::expected<std::vector<uint64_t>, ErrorCode> FileStorage::BatchGet(
}
}
uint64_t batch_id = allocated_batch->batch_id;
BatchGetResult batch_result{batch_id, allocated_batch->pointers};
MutexLocker locker(&client_buffer_mutex_);
client_buffer_allocated_batches_.emplace_back(std::move(allocated_batch));
client_buffer_allocated_batches_.emplace(batch_id,
std::move(allocated_batch));
auto end_time = std::chrono::steady_clock::now();
auto elapsed_time = std::chrono::duration_cast<std::chrono::microseconds>(
end_time - start_time)
.count();
VLOG(1) << "Time taken for FileStorage::BatchGet: " << elapsed_time
<< "us, key size: " << keys.size();
return allocate_res.value<>()->pointers;
<< "us, key size: " << keys.size() << ", batch_id: " << batch_id;
return batch_result;
}
tl::expected<void, ErrorCode> FileStorage::OffloadObjects(
@ -512,6 +499,7 @@ tl::expected<std::shared_ptr<FileStorage::AllocatedBatch>, ErrorCode>
FileStorage::AllocateBatch(const std::vector<std::string>& keys,
const std::vector<int64_t>& sizes) {
auto result = std::make_shared<AllocatedBatch>();
result->batch_id = next_batch_id_.fetch_add(1, std::memory_order_relaxed);
std::chrono::steady_clock::time_point now =
std::chrono::steady_clock::now();
auto lease_timeout =
@ -537,9 +525,9 @@ FileStorage::AllocateBatch(const std::vector<std::string>& keys,
{
MutexLocker locker(&client_buffer_mutex_);
auto gc_now = std::chrono::steady_clock::now();
auto it = client_buffer_allocated_batches_.begin();
while (it != client_buffer_allocated_batches_.end()) {
if (gc_now >= (*it)->lease_timeout) {
for (auto it = client_buffer_allocated_batches_.begin();
it != client_buffer_allocated_batches_.end();) {
if (gc_now >= it->second->lease_timeout) {
it = client_buffer_allocated_batches_.erase(it);
} else {
++it;
@ -582,14 +570,16 @@ void FileStorage::ClientBufferGCThreadFunc() {
MutexLocker locker(&client_buffer_mutex_);
if (!client_buffer_allocated_batches_.empty()) {
auto now = std::chrono::steady_clock::now();
client_buffer_allocated_batches_.erase(
std::remove_if(
client_buffer_allocated_batches_.begin(),
client_buffer_allocated_batches_.end(),
[&](const std::shared_ptr<AllocatedBatch>& batch) {
return now >= batch->lease_timeout;
}),
client_buffer_allocated_batches_.end());
for (auto it = client_buffer_allocated_batches_.begin();
it != client_buffer_allocated_batches_.end();) {
if (now >= it->second->lease_timeout) {
VLOG(1) << "GC releasing batch_id: " << it->first
<< " (lease expired)";
it = client_buffer_allocated_batches_.erase(it);
} else {
++it;
}
}
}
}
std::this_thread::sleep_for(
@ -598,4 +588,18 @@ void FileStorage::ClientBufferGCThreadFunc() {
LOG(INFO) << "action=client_buffer_gc_thread_stopped";
}
bool FileStorage::ReleaseBuffer(uint64_t batch_id) {
MutexLocker locker(&client_buffer_mutex_);
auto it = client_buffer_allocated_batches_.find(batch_id);
if (it != client_buffer_allocated_batches_.end()) {
VLOG(1) << "Releasing buffer for batch_id: " << batch_id
<< " (transfer completed)";
client_buffer_allocated_batches_.erase(it);
return true;
}
VLOG(1) << "batch_id " << batch_id
<< " not found (may have been GC'd already)";
return false;
}
} // namespace mooncake

65
mooncake-store/src/master_service.cpp
View File

@ -268,10 +268,11 @@ void MasterService::ClearInvalidHandles() {
while (it != shard->metadata.end()) {
if (CleanupStaleHandles(it->second)) {
// If the object is empty, we need to erase the iterator and
// also erase the key from processing_keys and
// replication_tasks.
// also erase the key from processing_keys,
// replication_tasks, and offloading_tasks.
shard->processing_keys.erase(it->first);
shard->replication_tasks.erase(it->first);
shard->offloading_tasks.erase(it->first);
it = shard->metadata.erase(it);
} else {
++it;
@ -796,10 +797,17 @@ auto MasterService::PutEnd(const UUID& client_id, const std::string& key,
[](Replica& replica) { replica.mark_complete(); });
if (enable_offload_) {
metadata.VisitReplicas(&Replica::fn_is_completed,
[this, &key](const Replica& replica) {
PushOffloadingQueue(key, replica);
});
auto& shard = accessor.GetShard();
metadata.VisitReplicas(
&Replica::fn_is_completed, [this, &key, &shard](Replica& replica) {
auto result = PushOffloadingQueue(key, replica);
if (result) {
replica.inc_refcnt();
shard->offloading_tasks.emplace(
key, OffloadingTask{replica.id(),
std::chrono::system_clock::now()});
}
});
}
// If the object is completed, remove it from the processing set.
@ -1642,6 +1650,26 @@ auto MasterService::NotifyOffloadSuccess(
for (size_t i = 0; i < keys.size(); ++i) {
const auto& key = keys[i];
const auto& metadata = metadatas[i];
// Release refcnt and clear offloading task.
{
MetadataAccessorRW accessor(this, key);
if (accessor.Exists()) {
auto& obj_metadata = accessor.Get();
auto& shard = accessor.GetShard();
auto task_it = shard->offloading_tasks.find(key);
if (task_it != shard->offloading_tasks.end()) {
auto source =
obj_metadata.GetReplicaByID(task_it->second.source_id);
if (source != nullptr) {
source->dec_refcnt();
}
shard->offloading_tasks.erase(task_it);
}
}
}
// Add LOCAL_DISK replica.
Replica replica(client_id, metadata.data_size,
metadata.transport_endpoint, ReplicaStatus::COMPLETE);
auto res = AddReplica(client_id, key, replica);
@ -1655,7 +1683,7 @@ auto MasterService::NotifyOffloadSuccess(
}
tl::expected<void, ErrorCode> MasterService::PushOffloadingQueue(
const std::string& key, const Replica& replica) {
const std::string& key, Replica& replica) {
const auto& segment_names = replica.get_segment_names();
if (segment_names.empty()) {
return {};
@ -1846,6 +1874,29 @@ void MasterService::DiscardExpiredProcessingReplicas(
task_it = shard->replication_tasks.erase(task_it);
}
// Part 3: Discard expired offloading operations.
for (auto task_it = shard->offloading_tasks.begin();
task_it != shard->offloading_tasks.end();) {
const auto ttl =
task_it->second.start_time + put_start_release_timeout_sec_;
if (ttl > now) {
task_it++;
continue;
}
auto metadata_it = shard->metadata.find(task_it->first);
if (metadata_it != shard->metadata.end()) {
auto source =
metadata_it->second.GetReplicaByID(task_it->second.source_id);
if (source != nullptr) {
source->dec_refcnt();
}
}
LOG(WARNING) << "Offloading task expired for key: " << task_it->first;
task_it = shard->offloading_tasks.erase(task_it);
}
if (!discarded_replicas.empty()) {
std::lock_guard lock(discarded_replicas_mutex_);
discarded_replicas_.splice(discarded_replicas_.end(),

38
mooncake-store/src/real_client.cpp
View File

@ -2656,10 +2656,20 @@ RealClient::batch_get_offload_object(const std::vector<std::string> &keys,
return tl::make_unexpected(result.error());
}
return BatchGetOffloadObjectResponse(
std::move(result.value()), client_->GetTransportEndpoint(),
result.value().batch_id, std::move(result.value().pointers),
client_->GetTransportEndpoint(),
file_storage_->config_.client_buffer_gc_ttl_ms);
}
bool RealClient::release_offload_buffer(uint64_t batch_id) {
if (!file_storage_) {
LOG(WARNING)
<< "release_offload_buffer called but file_storage_ is null";
return false;
}
return file_storage_->ReleaseBuffer(batch_id);
}
tl::expected<void, ErrorCode>
RealClient::batch_get_into_offload_object_internal(
const std::string &target_rpc_service_addr,
@ -2690,7 +2700,14 @@ RealClient::batch_get_into_offload_object_internal(
<< elapsed_time
<< "ms, with target_rpc_service_addr: " << target_rpc_service_addr
<< ", key size: " << objects.size()
<< "gc ttl: " << batchGetResp->gc_ttl_ms << "ms.";
<< ", batch_id: " << batchGetResp->batch_id
<< ", gc ttl: " << batchGetResp->gc_ttl_ms << "ms.";
// Release buffer immediately after transfer completion (fire-and-forget)
// This allows early buffer reclamation instead of waiting for GC lease
client_requester_->release_offload_buffer(target_rpc_service_addr,
batchGetResp->batch_id);
if (!result) {
LOG(ERROR) << "Batch get into offload object failed with error: "
<< result.error();
@ -2729,6 +2746,23 @@ ClientRequester::batch_get_offload_object(const std::string &client_addr,
return result;
}
void ClientRequester::release_offload_buffer(const std::string &client_addr,
uint64_t batch_id) {
// Fire-and-forget: attempt to release buffer, log errors but don't block
auto result = invoke_rpc<&RealClient::release_offload_buffer, bool>(
client_addr, batch_id);
if (!result) {
// This is expected in some cases (e.g., network issues, buffer already
// GC'd) Log at INFO level since GC will eventually clean up anyway
VLOG(1) << "Failed to release_offload_buffer for batch_id=" << batch_id
<< " at " << client_addr
<< " (will be GC'd): " << result.error();
} else {
VLOG(1) << "Successfully released buffer for batch_id=" << batch_id
<< " at " << client_addr;
}
}
template <auto ServiceMethod, typename ReturnType, typename... Args>
tl::expected<ReturnType, ErrorCode> ClientRequester::invoke_rpc(
const std::string &client_addr, Args &&...args) {

1
mooncake-store/src/real_client_main.cpp
View File

@ -56,6 +56,7 @@ void RegisterClientRpcService(coro_rpc::coro_rpc_server &server,
server.register_handler<&RealClient::query_task>(&real_client);
server.register_handler<&RealClient::batch_get_offload_object>(
&real_client);
server.register_handler<&RealClient::release_offload_buffer>(&real_client);
}
} // namespace mooncake

23
mooncake-store/src/storage_backend.cpp
View File

@ -757,6 +757,15 @@ std::unique_ptr<StorageFile> StorageBackend::create_file(
break;
}
#ifdef USE_URING
// Use O_DIRECT only for reads: write latency is not sensitive in this
// scenario, and O_DIRECT writes require 4096-byte alignment padding which
// corrupts meta file parsing and wastes disk space on data files.
if (use_uring_ && mode == FileMode::Read) {
flags |= O_DIRECT;
}
#endif
int fd = open(path.c_str(), flags | access_mode, 0644);
if (fd < 0) {
return nullptr;
@ -776,7 +785,11 @@ std::unique_ptr<StorageFile> StorageBackend::create_file(
#ifdef USE_URING
if (use_uring_) {
return std::make_unique<UringFile>(path, fd, 32, true);
// use_direct_io mirrors the O_DIRECT flag: true for reads, false for
// writes. This avoids unnecessary bounce-buffer allocation on the write
// path while keeping correct alignment enforcement on the read path.
bool use_direct_io = (mode == FileMode::Read);
return std::make_unique<UringFile>(path, fd, 32, use_direct_io);
}
#endif
return std::make_unique<PosixFile>(path, fd);
@ -2277,8 +2290,10 @@ BucketStorageBackend::OpenFile(const std::string& path, FileMode mode) const {
}
#ifdef USE_URING
// Add O_DIRECT flag when using uring for direct I/O
if (file_storage_config_.use_uring) {
// Use O_DIRECT only for reads: write latency is not sensitive in this
// scenario, and O_DIRECT writes require 4096-byte alignment padding which
// corrupts meta file parsing and wastes disk space on data files.
if (file_storage_config_.use_uring && mode == FileMode::Read) {
flags |= O_DIRECT;
}
#endif
@ -2290,7 +2305,7 @@ BucketStorageBackend::OpenFile(const std::string& path, FileMode mode) const {
return tl::make_unexpected(ErrorCode::FILE_OPEN_FAIL);
}
#ifdef USE_URING
if (file_storage_config_.use_uring) {
if (file_storage_config_.use_uring && mode == FileMode::Read) {
return std::make_unique<UringFile>(path, fd, 32, true);
}
#endif

1334
mooncake-store/src/uring_file.cpp
View File

File diff suppressed because it is too large Load Diff