From 6abce135f64770414e42d93dc0a95357d4313758 Mon Sep 17 00:00:00 2001 From: Cage Chen Date: Sat, 10 Oct 2026 13:23:45 +0800 Subject: [PATCH 1/4] [Bugfix][Store] Complete NoF tasks only after unlinking An error callback can finish a partially submitted task while it remains on the QoS chain. The submit loop then completes it again when unlinking. Require the task to be off the chain before publishing completion and reclaiming it. --- mooncake-store/src/transfer_task.cpp | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/mooncake-store/src/transfer_task.cpp b/mooncake-store/src/transfer_task.cpp index efa6d49a59..4459d61bde 100644 --- a/mooncake-store/src/transfer_task.cpp +++ b/mooncake-store/src/transfer_task.cpp @@ -80,13 +80,12 @@ static int CountSpdkNofQueuedTasks(const mooncake::SpdkNofTask* head) { } static inline void SpdkNofTaskCompletion(mooncake::SpdkNofTask* task) { - if (task->remaining_lba == 0 && task->outstanding_sub_io == 0) { + if (!task->on_chain && task->remaining_lba == 0 && + task->outstanding_sub_io == 0) { task->state->set_completed(task->failed ? mooncake::ErrorCode::TRANSFER_FAIL : mooncake::ErrorCode::OK); - if (!task->on_chain) { - delete task; - } + delete task; } } From c399ead1b6628721cfcb5f311b8ba03c5b26e0a1 Mon Sep 17 00:00:00 2001 From: Cage Chen Date: Mon, 12 Oct 2026 16:55:46 +0800 Subject: [PATCH 2/4] [Store] Add control task queues to NoF workers Add a per-worker control queue using the existing mutex and condition variable. Expose worker-local state through WorkerContext. Snapshot control and I/O queues under the same lock, then process control tasks before I/O tasks outside the lock. This prevents newly queued I/O from overtaking its preceding control task. Release callback captures outside the lock and drain queued control tasks before worker exit. Fail requests if task allocation fails after dequeueing. Signed-off-by: Cage Chen --- mooncake-store/include/transfer_task.h | 9 ++ mooncake-store/src/transfer_task.cpp | 141 +++++++++++------- mooncake-store/tests/CMakeLists.txt | 24 ++- .../tests/nof_worker_control_queue_test.cpp | 129 ++++++++++++++++ 4 files changed, 241 insertions(+), 62 deletions(-) create mode 100644 mooncake-store/tests/nof_worker_control_queue_test.cpp diff --git a/mooncake-store/include/transfer_task.h b/mooncake-store/include/transfer_task.h index 71f914fdfc..036a8c66d2 100644 --- a/mooncake-store/include/transfer_task.h +++ b/mooncake-store/include/transfer_task.h @@ -4,6 +4,7 @@ #include #include #include +#include #include #include #include @@ -493,6 +494,13 @@ struct SpdkNofQos { constexpr int kDefaultSpdkNofWorkers = 4; class SpdkNofWorkerPool { public: + struct WorkerContext; + using ControlTask = std::function; + + // Callbacks run on the target worker and must not wait for other tasks on + // that worker. Callers must stop posting before destroying the pool. + void PostControlTask(int worker_idx, ControlTask task); + explicit SpdkNofWorkerPool(int numa_socket_id = 0); ~SpdkNofWorkerPool(); @@ -515,6 +523,7 @@ class SpdkNofWorkerPool { int numa_socket_id_; std::vector workers_; std::unique_ptr[]> task_queue_; + std::unique_ptr[]> control_queue_; std::unique_ptr queue_mutex_; std::unique_ptr queue_cv_; std::atomic shutdown_; diff --git a/mooncake-store/src/transfer_task.cpp b/mooncake-store/src/transfer_task.cpp index 4459d61bde..bb5c28addb 100644 --- a/mooncake-store/src/transfer_task.cpp +++ b/mooncake-store/src/transfer_task.cpp @@ -333,6 +333,8 @@ SpdkNofWorkerPool::SpdkNofWorkerPool(int numa_socket_id) : worker_count_(NoFWorkerPoolConfig::AtFirstUse().worker_count), numa_socket_id_(numa_socket_id), task_queue_(std::make_unique[]>(worker_count_)), + control_queue_( + std::make_unique[]>(worker_count_)), queue_mutex_(std::make_unique(worker_count_)), queue_cv_(std::make_unique(worker_count_)), shutdown_(false) { @@ -364,6 +366,14 @@ SpdkNofWorkerPool::~SpdkNofWorkerPool() { VLOG(1) << "SpdkNofWorkerPool destroyed"; } +void SpdkNofWorkerPool::PostControlTask(int worker_idx, ControlTask task) { + { + std::lock_guard lock(queue_mutex_[worker_idx]); + control_queue_[worker_idx].push(std::move(task)); + } + queue_cv_[worker_idx].notify_one(); +} + void SpdkNofWorkerPool::submitTask(SpdkNofTask task) { if (!task.state) { LOG(ERROR) << "Attempting to submit spdk nof task without state"; @@ -441,81 +451,108 @@ static inline bool CheckSubTaskPool( return true; } +struct SpdkNofWorkerPool::WorkerContext { + int64_t total_outstanding_io{0}; + std::map> seg_to_qos; + std::stack sub_task_pool; + std::vector sub_task_chunks; + std::chrono::steady_clock::time_point last_debug_snapshot{ + std::chrono::steady_clock::now()}; + + ~WorkerContext() { + for (auto* sub_tasks : sub_task_chunks) delete[] sub_tasks; + } +}; + void SpdkNofWorkerPool::workerThread(int work_idx) { bindToSocket(numa_socket_id_); VLOG(2) << "SpdkNofWorkerPool worker thread started"; - int64_t total_outstanding_io = 0; - // std::set seg_set; - std::map> seg_to_qos; - std::stack sub_task_pool; - std::vector sub_task_chunks; + WorkerContext ctx; + auto& total_outstanding_io = ctx.total_outstanding_io; + auto& seg_to_qos = ctx.seg_to_qos; + auto& sub_task_pool = ctx.sub_task_pool; + auto& sub_task_chunks = ctx.sub_task_chunks; auto& task_queue = task_queue_[work_idx]; + auto& control_queue = control_queue_[work_idx]; auto& queue_cv = queue_cv_[work_idx]; auto& queue_mutex = queue_mutex_[work_idx]; - auto last_debug_snapshot = std::chrono::steady_clock::now(); + auto& last_debug_snapshot = ctx.last_debug_snapshot; if (!CheckSubTaskPool(sub_task_pool, sub_task_chunks, work_idx)) { return; } + std::queue controls; + std::queue incoming; while (true) { // Wait for task or shutdown signal { std::unique_lock lock(queue_mutex); - queue_cv.wait( - lock, [this, &task_queue, &total_outstanding_io, &seg_to_qos] { - return shutdown_.load() || !task_queue.empty() || - total_outstanding_io || HasBufferedTask(seg_to_qos); - }); + queue_cv.wait(lock, [this, &task_queue, &control_queue, + &total_outstanding_io, &seg_to_qos] { + return shutdown_.load() || !task_queue.empty() || + !control_queue.empty() || total_outstanding_io || + HasBufferedTask(seg_to_qos); + }); if (shutdown_.load() && task_queue.empty() && - (total_outstanding_io == 0) && !HasBufferedTask(seg_to_qos)) { + control_queue.empty() && (total_outstanding_io == 0) && + !HasBufferedTask(seg_to_qos)) { break; } - while (!task_queue.empty()) { - SpdkNofTask* task = new (std::nothrow) - SpdkNofTask(std::move(task_queue.front())); - if (task == nullptr) { + // Snapshot both queues together so new I/O cannot overtake a + // preceding control task while this worker executes callbacks. + controls.swap(control_queue); + incoming.swap(task_queue); + } + + while (!controls.empty()) { + controls.front()(ctx); + controls.pop(); + } + + while (!incoming.empty()) { + auto incoming_task = std::move(incoming.front()); + incoming.pop(); + SpdkNofTask* task = + new (std::nothrow) SpdkNofTask(std::move(incoming_task)); + if (task == nullptr) { + LOG(ERROR) << "alloc SpdkNofTask failed, worker " << work_idx; + incoming_task.state->set_completed(ErrorCode::TRANSFER_FAIL); + continue; + } + + SpdkNofQos* nof_qos = nullptr; + auto it = seg_to_qos.find(task->seg_handle); + if (it == seg_to_qos.end()) { + auto qos = std::make_unique( + SpdkWrapper::GetInstance().GetBlockSize(task->seg_handle)); + if (qos == nullptr) { LOG(ERROR) - << "alloc SpdkNofTask failed, worker " << work_idx; + << "alloc SpdkNofQos failed, worker " << work_idx; + delete task; continue; } - - SpdkNofQos* nof_qos = nullptr; - auto it = seg_to_qos.find(task->seg_handle); - if (it == seg_to_qos.end()) { - auto qos = std::make_unique( - SpdkWrapper::GetInstance().GetBlockSize( - task->seg_handle)); - if (qos == nullptr) { - LOG(ERROR) - << "alloc SpdkNofQos failed, worker " << work_idx; - delete task; - continue; - } - nof_qos = qos.get(); - seg_to_qos[task->seg_handle] = std::move(qos); - if (IsSpdkNofDebugEnabled()) { - LOG(INFO) - << "nof_qos_create worker_idx=" << work_idx - << " seg_handle=" << task->seg_handle - << " blocks_per_chunk=" - << seg_to_qos[task->seg_handle]->blocks_per_chunk - << " inflight_blocks_limit=" - << seg_to_qos[task->seg_handle] - ->inflight_blocks_limit; - } - } else { - nof_qos = it->second.get(); + nof_qos = qos.get(); + seg_to_qos[task->seg_handle] = std::move(qos); + if (IsSpdkNofDebugEnabled()) { + LOG(INFO) + << "nof_qos_create worker_idx=" << work_idx + << " seg_handle=" << task->seg_handle + << " blocks_per_chunk=" + << seg_to_qos[task->seg_handle]->blocks_per_chunk + << " inflight_blocks_limit=" + << seg_to_qos[task->seg_handle]->inflight_blocks_limit; } - task->io_count = &total_outstanding_io; - task->nof_qos = nof_qos; - task->on_chain = true; - nof_qos->PushTask(task); - task_queue.pop(); + } else { + nof_qos = it->second.get(); } + task->io_count = &total_outstanding_io; + task->nof_qos = nof_qos; + task->on_chain = true; + nof_qos->PushTask(task); } for (auto& [seg_handle, nof_qos] : seg_to_qos) { @@ -610,12 +647,6 @@ void SpdkNofWorkerPool::workerThread(int work_idx) { } } - for (auto* sub_tasks : sub_task_chunks) { - delete[] sub_tasks; - } - // seg_to_qos will automatically clean up SpdkNofQos objects using - // unique_ptr - VLOG(2) << "SpdkNofWorkerPool worker thread exiting"; } #endif diff --git a/mooncake-store/tests/CMakeLists.txt b/mooncake-store/tests/CMakeLists.txt index 08ed4dd305..9a33ceee82 100644 --- a/mooncake-store/tests/CMakeLists.txt +++ b/mooncake-store/tests/CMakeLists.txt @@ -200,15 +200,14 @@ if(USE_TENT) --gtest_filter=TransferTaskTest.TransferScatterHandlesFragmentedCpuBuffers ) set_tests_properties(transfer_scatter_tent_test PROPERTIES ENVIRONMENT - "MC_USE_TENT=1") + "MC_USE_TENT=1") add_test( NAME transfer_submitter_scatter_tent_test COMMAND transfer_task_test - --gtest_filter=TransferTaskTest.TransferSubmitterScatterUsesTentDataPath - ) - set_tests_properties(transfer_submitter_scatter_tent_test PROPERTIES - ENVIRONMENT "MC_USE_TENT=1") + --gtest_filter=TransferTaskTest.TransferSubmitterScatterUsesTentDataPath) + set_tests_properties(transfer_submitter_scatter_tent_test + PROPERTIES ENVIRONMENT "MC_USE_TENT=1") endif() add_store_test(tenant_quota_test tenant_quota_test.cpp) add_store_test(tenant_quota_ledger_test tenant_quota_ledger_test.cpp) @@ -247,8 +246,10 @@ add_store_test(cxl_segment_config_test cxl_segment_config_test.cpp) add_store_test(client_numa_config_test client_numa_config_test.cpp) add_store_test(local_hot_cache_config_test local_hot_cache_config_test.cpp) add_store_test(client_local_hot_cache_test client_local_hot_cache_test.cpp) -add_store_test(real_client_hugepage_register_test real_client_hugepage_register_test.cpp) -add_store_test(real_client_regular_page_register_test real_client_regular_page_register_test.cpp) +add_store_test(real_client_hugepage_register_test + real_client_hugepage_register_test.cpp) +add_store_test(real_client_regular_page_register_test + real_client_regular_page_register_test.cpp) add_store_test(client_tcp_local_memcpy_test client_tcp_local_memcpy_test.cpp) add_store_test(pybind_client_test pybind_client_test.cpp) add_store_test(standalone_client_test standalone_client_test.cpp) @@ -360,6 +361,15 @@ add_store_test(fileread_worker_pool_config_test fileread_worker_pool_config_test.cpp) add_store_test(nof_worker_pool_config_test nof_worker_pool_config_test.cpp) add_store_test(nof_debug_config_test nof_debug_config_test.cpp) + +if(USE_NOF) + add_store_test(nof_worker_control_queue_test + nof_worker_control_queue_test.cpp) + target_include_directories(nof_worker_control_queue_test + PRIVATE ${SPDK_INCLUDE_DIR} ${DPDK_INCLUDE_DIR}) + set_tests_properties(nof_worker_control_queue_test PROPERTIES TIMEOUT 30) +endif() + add_store_test(fileread_worker_pool_test fileread_worker_pool_test.cpp) if(MOONCAKE_OSS_ADAPTER_ENABLED) target_compile_definitions(file_storage_test PRIVATE HAVE_OSS_ADAPTER) diff --git a/mooncake-store/tests/nof_worker_control_queue_test.cpp b/mooncake-store/tests/nof_worker_control_queue_test.cpp new file mode 100644 index 0000000000..c8b5a6e8a1 --- /dev/null +++ b/mooncake-store/tests/nof_worker_control_queue_test.cpp @@ -0,0 +1,129 @@ +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "transfer_task.h" + +namespace mooncake { +namespace { +using namespace std::chrono_literals; +constexpr auto kTimeout = 5s; +using ThreadPair = std::pair; + +class NoFWorkerControlQueueTest : public ::testing::Test { + protected: + void SetUp() override { + google::InitGoogleLogging("NoFWorkerControlQueueTest"); + if (const char* value = std::getenv("MC_NOF_WORKERS")) + original_ = value; + ASSERT_EQ(setenv("MC_NOF_WORKERS", "2", 1), 0); + } + + void TearDown() override { + if (original_) + EXPECT_EQ(setenv("MC_NOF_WORKERS", original_->c_str(), 1), 0); + else + EXPECT_EQ(unsetenv("MC_NOF_WORKERS"), 0); + google::ShutdownGoogleLogging(); + } + + private: + std::optional original_; +}; + +TEST_F(NoFWorkerControlQueueTest, RunsOnSelectedWorkerInFifoOrderWithoutIo) { + constexpr int kTasks = 32; + std::array, 2> sequence; + std::array, 2> threads; + std::array, 2> done; + SpdkNofWorkerPool pool; + for (int index = 0; index < 2; ++index) { + for (int task = 0; task < kTasks; ++task) { + pool.PostControlTask(index, [&, index, task](auto&) { + sequence[index].push_back(task); + threads[index].push_back(std::this_thread::get_id()); + if (sequence[index].size() == kTasks) done[index].set_value(); + }); + } + } + for (auto& completion : done) + ASSERT_EQ(completion.get_future().wait_for(kTimeout), + std::future_status::ready); + + std::vector expected(kTasks); + std::iota(expected.begin(), expected.end(), 0); + for (int index = 0; index < 2; ++index) { + EXPECT_EQ(sequence[index], expected); + ASSERT_EQ(threads[index].size(), kTasks); + EXPECT_NE(threads[index][0], std::this_thread::get_id()); + for (auto thread : threads[index]) EXPECT_EQ(thread, threads[index][0]); + } + EXPECT_NE(threads[0][0], threads[1][0]); +} + +TEST_F(NoFWorkerControlQueueTest, CallbackCanPostToSameWorker) { + auto done = std::make_shared>(); + auto future = done->get_future(); + SpdkNofWorkerPool pool; + pool.PostControlTask(0, [&pool, done](auto&) { + auto outer = std::this_thread::get_id(); + pool.PostControlTask(0, [done, outer](auto&) { + done->set_value({outer, std::this_thread::get_id()}); + }); + }); + ASSERT_EQ(future.wait_for(kTimeout), std::future_status::ready); + auto [outer, inner] = future.get(); + EXPECT_EQ(outer, inner); + EXPECT_NE(inner, std::this_thread::get_id()); +} + +TEST_F(NoFWorkerControlQueueTest, ShutdownDrainsQueuedControlTasks) { + constexpr int kTasksPerWorker = 32; + std::atomic completed{0}; + std::array, 2> entered; + auto pool = std::make_unique(); + std::promise release; + auto gate = release.get_future().share(); + for (int index = 0; index < 2; ++index) { + pool->PostControlTask(index, [&, index, gate](auto&) { + entered[index].set_value(); + gate.wait(); + }); + } + for (auto& signal : entered) + ASSERT_EQ(signal.get_future().wait_for(kTimeout), + std::future_status::ready); + + for (int index = 0; index < 2; ++index) + for (int task = 0; task < kTasksPerWorker; ++task) + pool->PostControlTask(index, [&](auto&) { ++completed; }); + EXPECT_EQ(completed.load(), 0); + + std::promise stopping; + auto started = stopping.get_future(); + auto stopped = std::async(std::launch::async, + [owned = std::move(pool), &stopping]() mutable { + stopping.set_value(); + owned.reset(); + }); + EXPECT_EQ(started.wait_for(kTimeout), std::future_status::ready); + EXPECT_EQ(stopped.wait_for(20ms), std::future_status::timeout); + release.set_value(); + ASSERT_EQ(stopped.wait_for(kTimeout), std::future_status::ready); + stopped.get(); + EXPECT_EQ(completed.load(), 2 * kTasksPerWorker); +} +} // namespace +} // namespace mooncake From 92b539495d8ecd2ad3538f6d44f67f8b886a5692 Mon Sep 17 00:00:00 2001 From: Cage Chen Date: Wed, 14 Oct 2026 14:37:50 +0800 Subject: [PATCH 3/4] [Bugfix][Store] Recover NoF controllers after connection failures Detect connection failures through I/O and admin polling. Stop submission and drain qpairs on their owner workers before reconnecting the controller, then resume I/O after recovery. Bind segments during Open and let their owner workers create qpairs, advance asynchronous connections, and clean up on shutdown. Signed-off-by: Cage Chen --- .../benchmarks/nof_worker_pool_bench.cpp | 4 +- mooncake-store/include/spdk/spdk_wrapper.h | 44 ++- mooncake-store/include/transfer_task.h | 15 + mooncake-store/src/spdk/spdk_wrapper.cpp | 333 +++++++++++++++++- mooncake-store/src/transfer_task.cpp | 325 ++++++++++++----- mooncake-store/tests/e2e/CMakeLists.txt | 10 + .../tests/e2e/nof_controller_recovery_e2e.cpp | 176 +++++++++ mooncake-store/tests/e2e/nof_target.sh | 52 +++ mooncake-store/tests/e2e/readme.md | 53 +++ .../run_nof_controller_recovery_tcp_e2e.sh | 99 ++++++ .../tests/nof_worker_pool_config_test.cpp | 6 +- 11 files changed, 1011 insertions(+), 106 deletions(-) create mode 100644 mooncake-store/tests/e2e/nof_controller_recovery_e2e.cpp create mode 100644 mooncake-store/tests/e2e/nof_target.sh create mode 100755 mooncake-store/tests/e2e/run_nof_controller_recovery_tcp_e2e.sh diff --git a/mooncake-store/benchmarks/nof_worker_pool_bench.cpp b/mooncake-store/benchmarks/nof_worker_pool_bench.cpp index 6fbf27ef26..7a804d8512 100644 --- a/mooncake-store/benchmarks/nof_worker_pool_bench.cpp +++ b/mooncake-store/benchmarks/nof_worker_pool_bench.cpp @@ -731,12 +731,13 @@ int main(int argc, char **argv) { return 1; } + mooncake::SpdkNofWorkerPool pool; std::vector endpoints; endpoints.reserve(endpoint_strings.size()); std::unordered_set unique_handles; for (size_t i = 0; i < endpoint_strings.size(); ++i) { EndpointContext endpoint(endpoint_strings[i], FLAGS_seed + i); - endpoint.seg_handle = wrapper.OpenNofSegment(endpoint.endpoint); + endpoint.seg_handle = pool.OpenNofSegment(endpoint.endpoint); if (!endpoint.seg_handle) { LOG(ERROR) << "Failed to open NoF endpoint: " << endpoint.endpoint; @@ -837,7 +838,6 @@ int main(int argc, char **argv) { } std::vector endpoint_stats(endpoints.size()); - mooncake::SpdkNofWorkerPool pool; TimePoint start = Clock::now(); TimePoint warmup_end = start + std::chrono::seconds(FLAGS_warmup_sec); diff --git a/mooncake-store/include/spdk/spdk_wrapper.h b/mooncake-store/include/spdk/spdk_wrapper.h index e0ca16ab82..2c69fb78f8 100644 --- a/mooncake-store/include/spdk/spdk_wrapper.h +++ b/mooncake-store/include/spdk/spdk_wrapper.h @@ -2,11 +2,15 @@ #include #include #include +#include +#include #include #include #include #include #include +#include +#include #include #include #include @@ -22,6 +26,7 @@ constexpr int kSpdkNofOpNum = 2; struct nof_seg_handle; struct tr_info; struct ctrlr_info; +class SpdkNofWorkerPool; class SpdkWrapper { public: @@ -41,8 +46,10 @@ class SpdkWrapper { int64_t NvmePollProcessCompletion(nof_seg_handle *seg, uint32_t complete_per_seg); - /** @brief Open a NoF segment. */ - nof_seg_handle *OpenNofSegment(const std::string &tr_str); + // When recovery_enabled is true, only prepare the segment metadata; + // the owner worker creates the qpair. + nof_seg_handle *OpenNofSegment(const std::string &tr_str, + bool recovery_enabled = false); uint32_t GetBlockSize(const nof_seg_handle *seg_handle); @@ -99,6 +106,30 @@ class SpdkWrapper { */ static bool IsRegistrableRange(void *addr, size_t size); + // Non-owning identity; the pool unregisters its workers before destruction. + using WorkerId = std::pair; + // Dispatch through a registered callback to keep SpdkWrapper usable by + // Master without linking the client-side SpdkNofWorkerPool implementation. + using RecoveryTaskPoster = void (*)(WorkerId, ctrlr_info *, bool resume); + + // Register the client-side dispatcher before any worker is admitted. + // Repeated registrations must use the same function. + void RegisterRecoveryTaskPoster(RecoveryTaskPoster poster); + void StartControllerThread(); + + ctrlr_info *GetController(const nof_seg_handle *seg) const; + bool BindSegmentToWorker(nof_seg_handle *seg, WorkerId owner, + const std::function &enqueue); + + void RequestRecovery(ctrlr_info *ctrlr); + std::vector OwnedSegments(ctrlr_info *ctrlr, + WorkerId owner); + void OwnerDrained(ctrlr_info *ctrlr, WorkerId owner); + void UnregisterOwner(ctrlr_info *ctrlr, WorkerId owner); + + int SetupClientQpairIfNeeded(nof_seg_handle *seg); + bool DrainClientQpair(nof_seg_handle *seg); + private: struct ProbeBuffer { void *ptr{nullptr}; @@ -141,10 +172,19 @@ class SpdkWrapper { void ReplenishProbeRequestContextPoolLocked(size_t count); static void ProbeReadComplete(void *ctx, const struct spdk_nvme_cpl *cpl); + void ControllerLoop(); + std::atomic initialized{false}; std::mutex init_mutex; std::map> connected_ctrlrs; std::mutex ctrlrs_mutex; + + std::mutex controller_mutex_; + std::condition_variable controller_cv_; + bool controller_stop_{false}; + std::thread controller_thread_; + RecoveryTaskPoster post_recovery_task_{nullptr}; + std::map> probe_buffers_; std::mutex probe_buffers_mutex_; std::vector> probe_request_contexts_; diff --git a/mooncake-store/include/transfer_task.h b/mooncake-store/include/transfer_task.h index 036a8c66d2..c3c8df2adf 100644 --- a/mooncake-store/include/transfer_task.h +++ b/mooncake-store/include/transfer_task.h @@ -407,6 +407,7 @@ class MemcpyWorkerPool { #ifdef USE_NOF // struct SpdkNofSubTask; struct SpdkNofQos; +struct SpdkNofControllerState; /** * @brief Spdk nvmf operation descriptor @@ -454,6 +455,8 @@ struct SpdkNofSubTask { constexpr int kDefaultSpdkNofSubmitChunkBytes = (1 << 17); // 128k constexpr int kDefaultSpdkNofInflightBytesLimit = (1 << 25); // 32M struct SpdkNofQos { + SpdkNofControllerState* controller_state{nullptr}; + int inflight_blocks[kSpdkNofOpNum]; int blocks_per_chunk; int inflight_blocks_limit; @@ -485,6 +488,13 @@ struct SpdkNofQos { } }; +struct SpdkNofControllerState { + enum class Phase { Running, Cleaning, Paused }; + + Phase phase{Phase::Running}; + std::vector draining_segments; +}; + /** * @brief Thread pool for asynchronous spdk nvmf operations * @@ -510,6 +520,10 @@ class SpdkNofWorkerPool { SpdkNofWorkerPool(SpdkNofWorkerPool&&) = delete; SpdkNofWorkerPool& operator=(SpdkNofWorkerPool&&) = delete; + // Bind the segment to a worker before returning. That worker creates the + // qpair and starts the asynchronous connection. + nof_seg_handle* OpenNofSegment(const std::string& tr_str); + /** * @brief Submit a spdk nvmf task for async execution * @param task The spdk nvmf task to execute @@ -518,6 +532,7 @@ class SpdkNofWorkerPool { private: void workerThread(int work_idx); + void PostRecoveryTask(int worker_idx, ctrlr_info* ctrlr, bool resume); int worker_count_; int numa_socket_id_; diff --git a/mooncake-store/src/spdk/spdk_wrapper.cpp b/mooncake-store/src/spdk/spdk_wrapper.cpp index 5fbecf3e50..76c0043b68 100644 --- a/mooncake-store/src/spdk/spdk_wrapper.cpp +++ b/mooncake-store/src/spdk/spdk_wrapper.cpp @@ -6,6 +6,9 @@ #include #include #include +#include +#include +#include #if defined(__linux__) #include #include @@ -60,8 +63,14 @@ void ApplyCtrlrOptsFromEnv(struct spdk_nvme_ctrlr_opts *opts) { } // namespace struct nof_seg_handle { - struct spdk_nvme_qpair *qpair; - struct spdk_nvme_ns *ns; + struct spdk_nvme_qpair *qpair{nullptr}; + struct spdk_nvme_ns *ns{nullptr}; + + ctrlr_info *controller{nullptr}; + SpdkWrapper::WorkerId owner{nullptr, -1}; + + uint32_t block_size{0}; + bool disconnecting{false}; }; struct tr_info { @@ -71,7 +80,19 @@ struct tr_info { }; struct ctrlr_info { - struct spdk_nvme_ctrlr *ctrlr; + enum class State { Ready, Draining, Disconnecting, Reconnecting, Backoff }; + + struct spdk_nvme_ctrlr *ctrlr{nullptr}; + + // Fixed at creation, probe-only controllers leave recovery disabled. + bool recovery_enabled{false}; + State state{State::Ready}; + // One entry per owner worker, shared by its namespaces on this controller. + std::set owners; + // Owners still awaiting cleanup acknowledgement in the current recovery. + std::set pending_owners; + std::chrono::steady_clock::time_point retry_at{}; + std::map> ns_seg; std::mutex ns_mutex; }; @@ -141,6 +162,13 @@ bool SpdkWrapper::InitializeEnv() { } void SpdkWrapper::Cleanup() { + { + std::lock_guard lock(controller_mutex_); + controller_stop_ = true; + } + controller_cv_.notify_all(); + if (controller_thread_.joinable()) controller_thread_.join(); + if (initialized.load(std::memory_order_acquire)) { { std::lock_guard lock(ctrlrs_mutex); @@ -298,7 +326,9 @@ void SpdkWrapper::RecycleProbeRequestContext(ProbeRequestContext *ctx) { int64_t SpdkWrapper::NvmePollProcessCompletion(nof_seg_handle *seg, uint32_t complete_per_seg) { - return spdk_nvme_qpair_process_completions(seg->qpair, complete_per_seg); + return seg->qpair ? spdk_nvme_qpair_process_completions(seg->qpair, + complete_per_seg) + : 0; } int SpdkWrapper::ParseTransPortStr(const std::string &tr_str, tr_info *info) { @@ -367,7 +397,8 @@ int SpdkWrapper::ConnectController(const struct spdk_nvme_transport_id *trid, return spdk_nvme_probe(trid, (void *)info, probe_cb, attach_cb, NULL); } -nof_seg_handle *SpdkWrapper::OpenNofSegment(const std::string &tr_str) { +nof_seg_handle *SpdkWrapper::OpenNofSegment(const std::string &tr_str, + bool recovery_enabled) { // ConnectController -> spdk_nvme_probe() uses env-provided sockets and // memory, so the SPDK env must be up. Do not rely on an earlier // RegisterMemory() to have initialized it: callers now skip the @@ -393,6 +424,7 @@ nof_seg_handle *SpdkWrapper::OpenNofSegment(const std::string &tr_str) { if (it == connected_ctrlrs.end()) { auto new_info = std::make_unique(); info = new_info.get(); + info->recovery_enabled = recovery_enabled; ret = ConnectController(&tr.trid, info); if (ret != 0) { @@ -405,12 +437,22 @@ nof_seg_handle *SpdkWrapper::OpenNofSegment(const std::string &tr_str) { } } + // SPDK may reuse a controller for the same transport ID. Never enable + // recovery around a legacy probe connection that has no worker owner. + if (info->recovery_enabled != recovery_enabled) { + LOG(ERROR) << "Cannot mix probe and recoverable client ownership for " + << tr.ctrlr_key; + return nullptr; + } + nof_seg_handle *seg_handle = nullptr; struct spdk_nvme_qpair *qpair = nullptr; struct spdk_nvme_ns *ns = nullptr; { auto &ns_seg = info->ns_seg; std::lock_guard lock(info->ns_mutex); + if (info->state != ctrlr_info::State::Ready) return nullptr; + auto ns_it = ns_seg.find(tr.ns); if (ns_it != ns_seg.end()) { return ns_it->second.get(); @@ -423,15 +465,18 @@ nof_seg_handle *SpdkWrapper::OpenNofSegment(const std::string &tr_str) { return nullptr; } - qpair = spdk_nvme_ctrlr_alloc_io_qpair(info->ctrlr, nullptr, 0); - if (!qpair) { - LOG(ERROR) << "alloc spdk_nvme_qpair failed"; - return nullptr; - } - auto new_seg = std::make_unique(); + if (!recovery_enabled) { + qpair = spdk_nvme_ctrlr_alloc_io_qpair(info->ctrlr, nullptr, 0); + if (!qpair) { + LOG(ERROR) << "alloc spdk_nvme_qpair failed"; + return nullptr; + } + } new_seg->qpair = qpair; new_seg->ns = ns; + new_seg->controller = info; + new_seg->block_size = spdk_nvme_ns_get_sector_size(ns); seg_handle = new_seg.get(); ns_seg[tr.ns] = std::move(new_seg); } @@ -440,11 +485,12 @@ nof_seg_handle *SpdkWrapper::OpenNofSegment(const std::string &tr_str) { } uint32_t SpdkWrapper::GetBlockSize(const nof_seg_handle *seg_handle) { - if (!seg_handle || !seg_handle->ns) { - return INVALID_BLOCK_SIZE; + if (!seg_handle) return INVALID_BLOCK_SIZE; + if (!seg_handle->controller->recovery_enabled) { + return seg_handle->ns ? spdk_nvme_ns_get_sector_size(seg_handle->ns) + : INVALID_BLOCK_SIZE; } - - return spdk_nvme_ns_get_sector_size(seg_handle->ns); + return seg_handle->block_size; } int SpdkWrapper::SubmitRequest(const nof_seg_handle *seg_handle, void *ptr, @@ -564,4 +610,261 @@ bool SpdkWrapper::ProbeNofSegment(const std::string &tr_str, return ok; } +void SpdkWrapper::RegisterRecoveryTaskPoster(RecoveryTaskPoster poster) { + // Serialize registration from concurrently constructed worker pools. + std::lock_guard lock(controller_mutex_); + CHECK(poster != nullptr); + if (post_recovery_task_) { + CHECK(post_recovery_task_ == poster); + return; + } + post_recovery_task_ = poster; +} + +ctrlr_info *SpdkWrapper::GetController(const nof_seg_handle *seg) const { + return seg->controller; +} + +bool SpdkWrapper::BindSegmentToWorker(nof_seg_handle *seg, WorkerId owner, + const std::function &enqueue) { + auto *info = seg->controller; + + std::lock_guard lock(info->ns_mutex); + if (!info->recovery_enabled || info->state != ctrlr_info::State::Ready || + !seg->ns || seg->owner.first) + return false; + + bool owner_added = info->owners.insert(owner).second; + seg->owner = owner; + try { + enqueue(); + } catch (...) { + seg->owner = {nullptr, -1}; + if (owner_added) info->owners.erase(owner); + return false; + } + return true; +} + +void SpdkWrapper::RequestRecovery(ctrlr_info *info) { + std::lock_guard lock(info->ns_mutex); + if (!info->recovery_enabled || info->state != ctrlr_info::State::Ready) + return; + + info->state = ctrlr_info::State::Draining; + const auto *trid = spdk_nvme_ctrlr_get_transport_id(info->ctrlr); + LOG(WARNING) << "nof_recovery begin traddr=" << trid->traddr + << " trsvcid=" << trid->trsvcid << " subnqn=" << trid->subnqn + << " owners=" << info->owners.size(); + for (const auto &owner : info->owners) { + info->pending_owners.insert(owner); + post_recovery_task_(owner, info, false); + } + controller_cv_.notify_one(); +} + +std::vector SpdkWrapper::OwnedSegments(ctrlr_info *info, + WorkerId owner) { + std::lock_guard lock(info->ns_mutex); + std::vector result; + for (auto &[_, seg] : info->ns_seg) { + if (seg->owner == owner) result.push_back(seg.get()); + } + return result; +} + +void SpdkWrapper::OwnerDrained(ctrlr_info *info, WorkerId owner) { + std::lock_guard lock(info->ns_mutex); + info->pending_owners.erase(owner); + controller_cv_.notify_one(); +} + +void SpdkWrapper::UnregisterOwner(ctrlr_info *info, WorkerId owner) { + std::lock_guard lock(info->ns_mutex); + for (auto &[_, seg] : info->ns_seg) { + if (seg->owner == owner) { + assert(seg->qpair == nullptr); + seg->owner = {nullptr, -1}; + } + } + info->owners.erase(owner); + // Safe even if a new recovery added this owner: its qpairs are already + // freed. + info->pending_owners.erase(owner); + controller_cv_.notify_one(); +} + +int SpdkWrapper::SetupClientQpairIfNeeded(nof_seg_handle *seg) { + // Called only on its owner. The common case never takes a controller lock. + if (seg->qpair) return 0; + + auto *info = seg->controller; + { + std::lock_guard lock(info->ns_mutex); + if (info->state != ctrlr_info::State::Ready) return -EAGAIN; + if (!seg->ns) return -ENODEV; + } + + // This registered owner cannot acknowledge Cleanup while creating here. + // The cleanup barrier protects the controller without holding admission + // locked across a potentially slow transport connect. + spdk_nvme_io_qpair_opts opts; + spdk_nvme_ctrlr_get_default_io_qpair_opts(info->ctrlr, &opts, sizeof(opts)); + opts.async_mode = true; + opts.create_only = true; + seg->qpair = + spdk_nvme_ctrlr_alloc_io_qpair(info->ctrlr, &opts, sizeof(opts)); + if (!seg->qpair) return -ENOMEM; + + LOG(INFO) << "nof_qpair_async_connect_start ctrlr=" << info + << " nsid=" << spdk_nvme_ns_get_id(seg->ns) + << " qpair=" << seg->qpair << " pool=" << seg->owner.first + << " worker_idx=" << seg->owner.second; + + // Distinguish queue allocation pressure from a failed transport connect. + int rc = spdk_nvme_ctrlr_connect_io_qpair(info->ctrlr, seg->qpair); + return rc == 0 ? 0 : -ENXIO; +} + +bool SpdkWrapper::DrainClientQpair(nof_seg_handle *seg) { + if (!seg->qpair) return true; + + if (!seg->disconnecting) { + seg->disconnecting = true; + spdk_nvme_ctrlr_disconnect_io_qpair(seg->qpair); + } + + // In SPDK 23.01, explicit disconnect puts the qpair in DISCONNECTING. + // process_completions masks -ENXIO to 0 while still disconnecting, and + // returns -ENXIO after transport teardown. This is not a generic "error + // means drained" rule: it is used only after our explicit disconnect. + if (spdk_nvme_qpair_process_completions(seg->qpair, 0) != -ENXIO) + return false; + + spdk_nvme_ctrlr_free_io_qpair(seg->qpair); + seg->qpair = nullptr; + seg->disconnecting = false; + return true; +} + +void SpdkWrapper::StartControllerThread() { + std::lock_guard lock(controller_mutex_); + assert(post_recovery_task_ != nullptr); + if (!controller_thread_.joinable()) { + controller_stop_ = false; + controller_thread_ = std::thread(&SpdkWrapper::ControllerLoop, this); + } +} + +void SpdkWrapper::ControllerLoop() { + using State = ctrlr_info::State; + + while (true) { + bool stopping; + { + std::unique_lock lock(controller_mutex_); + controller_cv_.wait_for(lock, std::chrono::milliseconds(10)); + stopping = controller_stop_; + } + + std::vector controllers; + { + std::lock_guard lock(ctrlrs_mutex); + for (auto &[_, info] : connected_ctrlrs) { + if (info->recovery_enabled) controllers.push_back(info.get()); + } + } + + bool reconnect_in_progress = false; + for (auto *info : controllers) { + std::unique_lock lock(info->ns_mutex); + auto state = info->state; + + // SPDK holds its controller lock until reconnect succeeds or fails. + // Even during shutdown, keep polling on this thread until SPDK + // releases the lock; skip all other controller states. + if (stopping && state != State::Reconnecting) continue; + + if (state == State::Draining) { + if (!info->pending_owners.empty()) continue; + } else if (state == State::Backoff && + std::chrono::steady_clock::now() < info->retry_at) { + continue; + } + + lock.unlock(); + if (state == State::Ready) { + // In SPDK 23.01, admin polling fetches RDMA connection events + // and dispatches them to the affected qpairs. Keep polling even + // with keep-alive disabled so workers can detect remote + // disconnects. + if (spdk_nvme_ctrlr_process_admin_completions(info->ctrlr) < 0) + RequestRecovery(info); + // RequestRecovery() may have changed info->state to Draining, + // either here or on a worker. Skip writeback of the stale Ready + // snapshot. + continue; + } + + // Only this thread advances recovery after Ready -> Draining. + // Keep admission closed and publish the resulting state once. + if (state == State::Draining || state == State::Backoff) { + spdk_nvme_ctrlr_disconnect(info->ctrlr); + state = State::Disconnecting; + } + + if (state == State::Disconnecting && + spdk_nvme_ctrlr_process_admin_completions(info->ctrlr) == + -ENXIO) { + spdk_nvme_ctrlr_reconnect_async(info->ctrlr); + state = State::Reconnecting; + } + + int rc = 0; + if (state == State::Reconnecting) + rc = spdk_nvme_ctrlr_reconnect_poll_async(info->ctrlr); + + lock.lock(); + info->state = state; + if (state != State::Reconnecting) continue; + + if (rc == -EAGAIN) { + reconnect_in_progress = true; + continue; + } + + if (rc != 0) { + info->state = State::Backoff; + info->retry_at = + std::chrono::steady_clock::now() + std::chrono::seconds(1); + const auto *trid = + spdk_nvme_ctrlr_get_transport_id(info->ctrlr); + LOG(WARNING) << "nof_recovery retry traddr=" << trid->traddr + << " trsvcid=" << trid->trsvcid + << " subnqn=" << trid->subnqn << " rc=" << rc + << " retry_after_ms=1000"; + continue; + } + + for (auto &[nsid, seg] : info->ns_seg) { + seg->ns = nullptr; + if (!spdk_nvme_ctrlr_is_active_ns(info->ctrlr, nsid)) continue; + auto *ns = spdk_nvme_ctrlr_get_ns(info->ctrlr, nsid); + if (spdk_nvme_ns_get_sector_size(ns) == seg->block_size) + seg->ns = ns; + } + + for (const auto &owner : info->owners) + post_recovery_task_(owner, info, true); + info->state = State::Ready; + const auto *trid = spdk_nvme_ctrlr_get_transport_id(info->ctrlr); + LOG(INFO) << "nof_recovery ready traddr=" << trid->traddr + << " trsvcid=" << trid->trsvcid + << " subnqn=" << trid->subnqn; + } + + if (stopping && !reconnect_in_progress) return; + } +} + } // namespace mooncake diff --git a/mooncake-store/src/transfer_task.cpp b/mooncake-store/src/transfer_task.cpp index bb5c28addb..06cd13f8d4 100644 --- a/mooncake-store/src/transfer_task.cpp +++ b/mooncake-store/src/transfer_task.cpp @@ -341,6 +341,12 @@ SpdkNofWorkerPool::SpdkNofWorkerPool(int numa_socket_id) VLOG(1) << "Creating SpdkNofWorkerPool with " << worker_count_ << " workers"; + SpdkWrapper::GetInstance().RegisterRecoveryTaskPoster( + [](SpdkWrapper::WorkerId owner, ctrlr_info* ctrlr, bool resume) { + owner.first->PostRecoveryTask(owner.second, ctrlr, resume); + }); + SpdkWrapper::GetInstance().StartControllerThread(); + // Start worker threads workers_.reserve(worker_count_); for (int i = 0; i < worker_count_; ++i) { @@ -349,11 +355,14 @@ SpdkNofWorkerPool::SpdkNofWorkerPool(int numa_socket_id) } SpdkNofWorkerPool::~SpdkNofWorkerPool() { - if (shutdown_.exchange(true)) { - return; + { + std::lock_guard lock(seg_mutex_); + if (shutdown_.exchange(true)) return; } for (int i = 0; i < worker_count_; ++i) { + // Take the queue lock so shutdown cannot miss a worker entering wait. + std::lock_guard lock(queue_mutex_[i]); queue_cv_[i].notify_all(); } @@ -380,6 +389,7 @@ void SpdkNofWorkerPool::submitTask(SpdkNofTask task) { return; } + std::lock_guard lock(seg_mutex_); if (shutdown_.load()) { LOG(WARNING) << "Attempting to submit task to shutdown SpdkNofWorkerPool"; @@ -387,31 +397,15 @@ void SpdkNofWorkerPool::submitTask(SpdkNofTask task) { return; } - int worker_idx = -1; - { - std::lock_guard lock(seg_mutex_); - nof_seg_handle* seg = task.seg_handle; - bool new_binding = false; - if (seg_to_worker_.find(seg) != seg_to_worker_.end()) { - worker_idx = seg_to_worker_[seg]; - } else { - worker_idx = (seg_num++ % worker_count_); - seg_to_worker_[seg] = worker_idx; - new_binding = true; - } - if (new_binding && IsSpdkNofDebugEnabled()) { - LOG(INFO) << "nof_worker_bind seg_handle=" << seg - << " worker_idx=" << worker_idx; - } - } - if (worker_idx < 0 || worker_idx >= worker_count_) { - LOG(ERROR) << "seg is not bind to invalid worker " << worker_idx; + auto it = seg_to_worker_.find(task.seg_handle); + if (it == seg_to_worker_.end()) { + LOG(ERROR) << "NoF segment is not bound to this worker pool"; task.state->set_completed(ErrorCode::TRANSFER_FAIL); return; } - + int worker_idx = it->second; { - std::lock_guard lock(queue_mutex_[worker_idx]); + std::lock_guard queue_lock(queue_mutex_[worker_idx]); task_queue_[worker_idx].push(std::move(task)); } queue_cv_[worker_idx].notify_one(); @@ -452,23 +446,102 @@ static inline bool CheckSubTaskPool( } struct SpdkNofWorkerPool::WorkerContext { + using Phase = SpdkNofControllerState::Phase; + + SpdkNofWorkerPool& pool; + int index; + SpdkWrapper& wrapper{SpdkWrapper::GetInstance()}; + int64_t total_outstanding_io{0}; + std::map controllers; std::map> seg_to_qos; + std::stack sub_task_pool; std::vector sub_task_chunks; std::chrono::steady_clock::time_point last_debug_snapshot{ std::chrono::steady_clock::now()}; + WorkerContext(SpdkNofWorkerPool& p, int i) : pool(p), index(i) {} + ~WorkerContext() { for (auto* sub_tasks : sub_task_chunks) delete[] sub_tasks; } + + SpdkWrapper::WorkerId OwnerId() { return {&pool, index}; } + + bool HasWork() const { + if (total_outstanding_io || HasBufferedTask(seg_to_qos)) return true; + for (auto& [_, state] : controllers) + if (state.phase == Phase::Cleaning) return true; + return false; + } + + static void FailQueued(SpdkNofQos& qos) { + for (int op = 0; op < kSpdkNofOpNum; ++op) { + while (qos.head[op]) { + auto* task = qos.head[op]; + qos.PopTask(op); + task->on_chain = false; + task->remaining_lba = 0; + task->failed = true; + SpdkNofTaskCompletion(task); + } + } + } + + void BeginControllerCleanup(ctrlr_info* ctrlr) { + auto& state = controllers[ctrlr]; + if (state.phase == Phase::Cleaning) return; + if (state.phase == Phase::Paused && !pool.shutdown_.load()) return; + + // On shutdown, conservatively rescan owned segments before + // UnregisterOwner(), even if the previous cleanup reached Paused. + state.phase = Phase::Cleaning; + state.draining_segments = wrapper.OwnedSegments(ctrlr, OwnerId()); + + for (auto& [_, qos] : seg_to_qos) + if (qos->controller_state == &state) FailQueued(*qos); + } + + void AdvanceCleanup() { + for (auto& [ctrlr, state] : controllers) { + if (state.phase != Phase::Cleaning) continue; + + bool drained = true; + for (auto* seg : state.draining_segments) + drained &= wrapper.DrainClientQpair(seg); + if (!drained) continue; + + for (auto& [_, qos] : seg_to_qos) { + if (qos->controller_state == &state) { + assert(qos->Empty()); + assert(qos->inflight_blocks[kSpdkNofOpRead] == 0 && + qos->inflight_blocks[kSpdkNofOpWrite] == 0); + } + } + + state.draining_segments.clear(); + state.phase = Phase::Paused; + wrapper.OwnerDrained(ctrlr, OwnerId()); + } + } + + void ResumeController(ctrlr_info* ctrlr) { + if (pool.shutdown_.load()) return; + + auto it = controllers.find(ctrlr); + assert(it != controllers.end()); + assert(it->second.phase == Phase::Paused); + it->second.phase = Phase::Running; + } }; void SpdkNofWorkerPool::workerThread(int work_idx) { bindToSocket(numa_socket_id_); VLOG(2) << "SpdkNofWorkerPool worker thread started"; - WorkerContext ctx; + using Phase = SpdkNofControllerState::Phase; + WorkerContext ctx(*this, work_idx); auto& total_outstanding_io = ctx.total_outstanding_io; auto& seg_to_qos = ctx.seg_to_qos; auto& sub_task_pool = ctx.sub_task_pool; @@ -478,32 +551,23 @@ void SpdkNofWorkerPool::workerThread(int work_idx) { auto& queue_cv = queue_cv_[work_idx]; auto& queue_mutex = queue_mutex_[work_idx]; auto& last_debug_snapshot = ctx.last_debug_snapshot; - - if (!CheckSubTaskPool(sub_task_pool, sub_task_chunks, work_idx)) { - return; - } - + auto& wrapper = ctx.wrapper; std::queue controls; std::queue incoming; + while (true) { - // Wait for task or shutdown signal { std::unique_lock lock(queue_mutex); - queue_cv.wait(lock, [this, &task_queue, &control_queue, - &total_outstanding_io, &seg_to_qos] { - return shutdown_.load() || !task_queue.empty() || - !control_queue.empty() || total_outstanding_io || - HasBufferedTask(seg_to_qos); - }); - - if (shutdown_.load() && task_queue.empty() && - control_queue.empty() && (total_outstanding_io == 0) && - !HasBufferedTask(seg_to_qos)) { - break; - } - - // Snapshot both queues together so new I/O cannot overtake a - // preceding control task while this worker executes callbacks. + auto ready = [this, &control_queue, &task_queue, &ctx] { + return shutdown_.load() || !control_queue.empty() || + !task_queue.empty() || ctx.HasWork(); + }; + // Poll idle qpairs every 1 ms to advance async connections + // and detect reported transport errors. + if (seg_to_qos.empty()) + queue_cv.wait(lock, ready); + else + queue_cv.wait_for(lock, std::chrono::milliseconds(1), ready); controls.swap(control_queue); incoming.swap(task_queue); } @@ -513,49 +577,49 @@ void SpdkNofWorkerPool::workerThread(int work_idx) { controls.pop(); } + // Each request's registration task has already run, + // so its worker-local QoS entry must exist. while (!incoming.empty()) { auto incoming_task = std::move(incoming.front()); incoming.pop(); - SpdkNofTask* task = + auto& qos = seg_to_qos.at(incoming_task.seg_handle); + if (qos->controller_state->phase != Phase::Running || + shutdown_.load()) { + incoming_task.state->set_completed(ErrorCode::TRANSFER_FAIL); + continue; + } + auto* task = new (std::nothrow) SpdkNofTask(std::move(incoming_task)); - if (task == nullptr) { + if (!task) { LOG(ERROR) << "alloc SpdkNofTask failed, worker " << work_idx; incoming_task.state->set_completed(ErrorCode::TRANSFER_FAIL); continue; } - SpdkNofQos* nof_qos = nullptr; - auto it = seg_to_qos.find(task->seg_handle); - if (it == seg_to_qos.end()) { - auto qos = std::make_unique( - SpdkWrapper::GetInstance().GetBlockSize(task->seg_handle)); - if (qos == nullptr) { - LOG(ERROR) - << "alloc SpdkNofQos failed, worker " << work_idx; - delete task; - continue; - } - nof_qos = qos.get(); - seg_to_qos[task->seg_handle] = std::move(qos); - if (IsSpdkNofDebugEnabled()) { - LOG(INFO) - << "nof_qos_create worker_idx=" << work_idx - << " seg_handle=" << task->seg_handle - << " blocks_per_chunk=" - << seg_to_qos[task->seg_handle]->blocks_per_chunk - << " inflight_blocks_limit=" - << seg_to_qos[task->seg_handle]->inflight_blocks_limit; - } - } else { - nof_qos = it->second.get(); - } task->io_count = &total_outstanding_io; - task->nof_qos = nof_qos; + task->nof_qos = qos.get(); task->on_chain = true; - nof_qos->PushTask(task); + qos->PushTask(task); + } + + // Use one shutdown snapshot for cleanup and exit in this iteration. + const bool stopping = shutdown_.load(); + if (stopping) { + for (auto& [ctrlr, _] : ctx.controllers) + ctx.BeginControllerCleanup(ctrlr); } for (auto& [seg_handle, nof_qos] : seg_to_qos) { + if (nof_qos->controller_state->phase != Phase::Running || + nof_qos->Empty()) + continue; + int connect_rc = wrapper.SetupClientQpairIfNeeded(seg_handle); + if (connect_rc != 0) { + if (connect_rc == -ENXIO) + wrapper.RequestRecovery(wrapper.GetController(seg_handle)); + ctx.FailQueued(*nof_qos); + continue; + } uint32_t block_size = SpdkWrapper::GetInstance().GetBlockSize(seg_handle); for (int i = 0; i < kSpdkNofOpNum; ++i) { @@ -591,7 +655,13 @@ void SpdkNofWorkerPool::workerThread(int work_idx) { sub_task); if (ret != 0) { LOG(ERROR) << "work " << work_idx << ", seg " - << task->seg_handle << " submit io fail"; + << task->seg_handle << " submit io fail" + << ", ret " << ret << ", op " << task->op + << ", lba " << submit_lba + << ", lba_count " << submit_lba_count; + if (ret == -ENXIO) + wrapper.RequestRecovery( + wrapper.GetController(seg_handle)); task->failed = true; task->remaining_lba = 0; } else { @@ -612,17 +682,17 @@ void SpdkNofWorkerPool::workerThread(int work_idx) { } } - if (total_outstanding_io > 0) { - int64_t ret = 0; - for (auto& it : seg_to_qos) { - ret = SpdkWrapper::GetInstance().NvmePollProcessCompletion( - it.first, 0); - if (ret < 0) { - LOG(ERROR) << "poll completion error: ret " << ret; - } + for (auto& [seg, qos] : seg_to_qos) { + if (qos->controller_state->phase != Phase::Running) continue; + int64_t rc = wrapper.NvmePollProcessCompletion(seg, 0); + if (rc < 0) { + LOG(ERROR) << "poll completion error: ret " << rc; + wrapper.RequestRecovery(wrapper.GetController(seg)); } } + ctx.AdvanceCleanup(); + if (IsSpdkNofDebugEnabled()) { auto now = std::chrono::steady_clock::now(); auto elapsed = @@ -645,10 +715,97 @@ void SpdkNofWorkerPool::workerThread(int work_idx) { last_debug_snapshot = now; } } + + if (stopping && !ctx.HasWork()) { + // Unregister under the controller's publication lock. No further + // recovery tasks can reference this worker after these calls. + for (auto& [ctrlr, _] : ctx.controllers) + ctx.wrapper.UnregisterOwner(ctrlr, ctx.OwnerId()); + std::lock_guard lock(queue_mutex); + if (control_queue.empty() && task_queue.empty()) break; + } } VLOG(2) << "SpdkNofWorkerPool worker thread exiting"; } + +nof_seg_handle* SpdkNofWorkerPool::OpenNofSegment(const std::string& tr_str) { + if (shutdown_.load()) return nullptr; + auto& wrapper = SpdkWrapper::GetInstance(); + auto* seg = wrapper.OpenNofSegment(tr_str, true); + if (!seg) return nullptr; + + std::lock_guard lock(seg_mutex_); + if (shutdown_.load()) return nullptr; + if (seg_to_worker_.find(seg) != seg_to_worker_.end()) return seg; + + int worker_idx = seg_num % worker_count_; + seg_to_worker_.emplace(seg, worker_idx); + try { + if (!wrapper.BindSegmentToWorker(seg, {this, worker_idx}, [&] { + PostControlTask( + worker_idx, [seg, worker_idx](WorkerContext& ctx) { + auto* ctrlr = ctx.wrapper.GetController(seg); + auto& state = ctx.controllers[ctrlr]; + auto qos = std::make_unique( + ctx.wrapper.GetBlockSize(seg)); + qos->controller_state = &state; + if (IsSpdkNofDebugEnabled()) { + LOG(INFO) + << "nof_qos_create worker_idx=" << worker_idx + << " seg_handle=" << seg + << " blocks_per_chunk=" << qos->blocks_per_chunk + << " inflight_blocks_limit=" + << qos->inflight_blocks_limit; + } + ctx.seg_to_qos.emplace(seg, std::move(qos)); + // SpdkWrapper::BindSegmentToWorker() published this + // owner before PostControlTask() queued this task. + // Shutdown may start before the task executes. + // SpdkNofWorkerPool::workerThread() traverses + // ctx.controllers and calls + // SpdkWrapper::UnregisterOwner() before exiting. Keep + // the local registration above so it finds this + // controller; otherwise the controller could retain an + // owner referencing a destroyed pool. Skip only + // SpdkWrapper::SetupClientQpairIfNeeded() below. + if (ctx.pool.shutdown_.load() || + state.phase != WorkerContext::Phase::Running) + return; + + int rc = ctx.wrapper.SetupClientQpairIfNeeded(seg); + if (rc == -ENXIO) ctx.wrapper.RequestRecovery(ctrlr); + if (rc != 0 && rc != -EAGAIN) { + LOG(WARNING) + << "NoF initial qpair setup failed: seg=" << seg + << ", ret=" << rc; + } + }); + })) { + seg_to_worker_.erase(seg); + return nullptr; + } + } catch (...) { + seg_to_worker_.erase(seg); + return nullptr; + } + ++seg_num; + if (IsSpdkNofDebugEnabled()) { + LOG(INFO) << "nof_worker_bind seg_handle=" << seg + << " worker_idx=" << worker_idx; + } + return seg; +} + +void SpdkNofWorkerPool::PostRecoveryTask(int worker_idx, ctrlr_info* ctrlr, + bool resume) { + PostControlTask(worker_idx, [ctrlr, resume](WorkerContext& ctx) { + if (resume) + ctx.ResumeController(ctrlr); + else + ctx.BeginControllerCleanup(ctrlr); + }); +} #endif // ============================================================================ @@ -1995,7 +2152,7 @@ std::optional TransferSubmitter::submitSpdkNofOperation( } nof_seg_handle* seg_handle = - SpdkWrapper::GetInstance().OpenNofSegment(handle.transport_endpoint_); + spdk_nvmf_pool_->OpenNofSegment(handle.transport_endpoint_); if (!seg_handle) { LOG(ERROR) << "Failed to open NoF segment endpoint=" << handle.transport_endpoint_; diff --git a/mooncake-store/tests/e2e/CMakeLists.txt b/mooncake-store/tests/e2e/CMakeLists.txt index 1ff8ac7f9a..bb3f50faa3 100644 --- a/mooncake-store/tests/e2e/CMakeLists.txt +++ b/mooncake-store/tests/e2e/CMakeLists.txt @@ -96,3 +96,13 @@ add_executable(oplog_batch_e2e_test oplog_batch_e2e_test.cpp process_handler.cpp target_link_libraries(oplog_batch_e2e_test PUBLIC glog::glog gtest pthread) add_test(NAME oplog_batch_e2e_test COMMAND oplog_batch_e2e_test) set_tests_properties(oplog_batch_e2e_test PROPERTIES LABELS oplog_batch_smoke) + +# Run through run_nof_controller_recovery_tcp_e2e.sh, which manages the target. +if(USE_NOF) + add_executable(nof_controller_recovery_e2e nof_controller_recovery_e2e.cpp) + target_include_directories(nof_controller_recovery_e2e + PRIVATE ${SPDK_INCLUDE_DIR} ${DPDK_INCLUDE_DIR}) + target_link_libraries( + nof_controller_recovery_e2e PRIVATE mooncake_store gflags::gflags + glog::glog pthread) +endif() diff --git a/mooncake-store/tests/e2e/nof_controller_recovery_e2e.cpp b/mooncake-store/tests/e2e/nof_controller_recovery_e2e.cpp new file mode 100644 index 0000000000..0416a68211 --- /dev/null +++ b/mooncake-store/tests/e2e/nof_controller_recovery_e2e.cpp @@ -0,0 +1,176 @@ +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "transfer_task.h" + +DEFINE_string(endpoint, "", "TCP controller endpoint without a namespace ID"); +DEFINE_string(control_dir, "", + "Target lifecycle markers written by the runner"); +DEFINE_uint32(timeout_sec, 30, "Timeout for each test phase"); + +namespace { +using namespace mooncake; +using namespace std::chrono_literals; +using Clock = std::chrono::steady_clock; +constexpr size_t kIoSize = 256 * 1024; +using Segments = std::array; +using Buffer = std::unique_ptr; +using Buffers = std::array; +using Requests = std::array, 2>; + +void Require(bool ok, const std::string& message) { + if (!ok) throw std::runtime_error(message); +} + +Clock::time_point Deadline() { + return Clock::now() + std::chrono::seconds(FLAGS_timeout_sec); +} + +void WaitForMarker(const char* name) { + auto deadline = Deadline(); + auto path = std::filesystem::path(FLAGS_control_dir) / name; + while (!std::filesystem::exists(path)) { + Require(Clock::now() < deadline, + "Timed out waiting for " + path.string()); + std::this_thread::sleep_for(1ms); + } +} + +std::thread::id WorkerBarrier(SpdkNofWorkerPool& pool, int index) { + auto done = std::make_shared>(); + auto future = done->get_future(); + pool.PostControlTask( + index, [done](auto&) { done->set_value(std::this_thread::get_id()); }); + Require(future.wait_until(Deadline()) == std::future_status::ready, + "Worker barrier timed out"); + return future.get(); +} + +Requests Submit(SpdkNofWorkerPool& pool, const Segments& segments, + const Buffers& buffers, int op) { + Requests requests; + for (size_t i = 0; i < segments.size(); ++i) { + auto block_size = SpdkWrapper::GetInstance().GetBlockSize(segments[i]); + requests[i] = std::make_shared(); + pool.submitTask(SpdkNofTask(segments[i], buffers[i].get(), 0, + kIoSize / block_size, op, requests[i])); + } + return requests; +} + +bool Wait(const Requests& requests, Clock::time_point deadline) { + bool success = true; + for (const auto& request : requests) { + while (!request->is_completed()) { + Require(Clock::now() < deadline, "I/O did not complete"); + std::this_thread::sleep_for(1ms); + } + success &= request->get_result() == ErrorCode::OK; + } + return success; +} + +bool WriteRead(SpdkNofWorkerPool& pool, const Segments& segments, + const Buffers& buffers, int pattern, + Clock::time_point deadline) { + for (size_t i = 0; i < buffers.size(); ++i) + std::memset(buffers[i].get(), pattern + i, kIoSize); + if (!Wait(Submit(pool, segments, buffers, kSpdkNofOpWrite), deadline)) + return false; + for (const auto& buffer : buffers) std::memset(buffer.get(), 0, kIoSize); + if (!Wait(Submit(pool, segments, buffers, kSpdkNofOpRead), deadline)) + return false; + for (size_t i = 0; i < buffers.size(); ++i) { + auto* bytes = static_cast(buffers[i].get()); + for (size_t offset = 0; offset < kIoSize; ++offset) + Require(bytes[offset] == pattern + i, + "Data mismatch in namespace " + std::to_string(i + 1)); + } + return true; +} + +void Run() { + Require(!FLAGS_endpoint.empty() && !FLAGS_control_dir.empty() && + FLAGS_timeout_sec > 0, + "Specify --endpoint, --control_dir and a positive --timeout_sec"); + Require(setenv("MC_NOF_WORKERS", "2", 1) == 0, "setenv failed"); + auto& wrapper = SpdkWrapper::GetInstance(); + // Buffers outlive the pool, including when a failed check unwinds Run(). + Buffers buffers{Buffer(wrapper.Alloc(kIoSize, 4096, -1), spdk_free), + Buffer(wrapper.Alloc(kIoSize, 4096, -1), spdk_free)}; + Require(buffers[0] && buffers[1], "DMA buffer allocation failed"); + SpdkNofWorkerPool pool; + Segments segments{}; + for (size_t i = 0; i < segments.size(); ++i) { + segments[i] = pool.OpenNofSegment(FLAGS_endpoint + + " ns:" + std::to_string(i + 1)); + Require(segments[i] != nullptr, "Open failed"); + auto block_size = wrapper.GetBlockSize(segments[i]); + Require(block_size != 0 && block_size != INVALID_BLOCK_SIZE && + kIoSize % block_size == 0, + "Invalid namespace block size"); + } + Require(wrapper.GetController(segments[0]) == + wrapper.GetController(segments[1]), + "Namespaces must share one controller"); + auto first = WorkerBarrier(pool, 0); + Require(first != WorkerBarrier(pool, 1), "Expected two distinct workers"); + Require(WriteRead(pool, segments, buffers, 0x31, Deadline()), + "Initial read/write failed"); + std::cout << "PASS initial_io" << std::endl; + + // The runner freezes the target before these requests are issued, so it + // cannot race to complete them before the fault. Neither pool nor handles + // are recreated during the test. + WaitForMarker("target-paused"); + auto pending = Submit(pool, segments, buffers, kSpdkNofOpWrite); + for (int round = 0; round < 2; ++round) + for (int index = 0; index < 2; ++index) WorkerBarrier(pool, index); + // Control tasks precede I/O within a snapshot. Two barriers ensure the + // worker has processed the request snapshot before we announce the fault. + for (const auto& request : pending) + Require(!request->is_completed(), "Target completed I/O while paused"); + std::cout << "READY fault_io_pending" << std::endl; + + Wait(pending, Deadline()); + for (const auto& request : pending) + Require(request->get_result() == ErrorCode::TRANSFER_FAIL, + "Disconnected I/O did not fail"); + std::cout << "PASS failed_io_drained" << std::endl; + + WaitForMarker("target-restarted"); + auto deadline = Deadline(); + // Malloc bdev data is lost on restart: write a new pattern before reading. + while (!WriteRead(pool, segments, buffers, 0x73, deadline)) { + Require(Clock::now() < deadline, "Controller did not recover"); + std::this_thread::sleep_for(10ms); + } +} +} // namespace + +int main(int argc, char** argv) { + gflags::ParseCommandLineFlags(&argc, &argv, true); + google::InitGoogleLogging(argv[0]); + FLAGS_logtostderr = true; + try { + Run(); + std::cout << "PASS controller_recovery" << std::endl; + return 0; + } catch (const std::exception& error) { + std::cerr << "FAIL controller_recovery: " << error.what() << std::endl; + return 1; + } +} diff --git a/mooncake-store/tests/e2e/nof_target.sh b/mooncake-store/tests/e2e/nof_target.sh new file mode 100644 index 0000000000..d455a89092 --- /dev/null +++ b/mooncake-store/tests/e2e/nof_target.sh @@ -0,0 +1,52 @@ +#!/usr/bin/env bash +# Source from a runner that owns the EXIT trap. Configure SPDK_DIR, RPC_SOCKET, +# TARGET_NQN, TARGET_HOST and TARGET_PORT before starting a target. +# TARGET_PID remains available to the runner for fault injection and cleanup. + +nof_target_rpc() { + python3 "$SPDK_DIR/scripts/rpc.py" -s "$RPC_SOCKET" -t 2 "$@" +} + +nof_target_start() { + local log=$1 deadline=$((SECONDS + ${TIMEOUT_SEC:-30})) + [[ -z "${TARGET_PID:-}" ]] || { echo "NoF target already started" >&2; return 1; } + [[ ! -e "$RPC_SOCKET" && ! -e "$RPC_SOCKET.lock" ]] || { + echo "NoF target socket already exists: $RPC_SOCKET" >&2 + return 1 + } + NOF_TARGET_SETUP_LOG="${log%.log}-setup.log" + "$SPDK_DIR/build/bin/nvmf_tgt" -m "${TARGET_CORE_MASK:-0x1}" -u \ + -s "${TARGET_MEM_MB:-512}" --iova-mode=va --disable-cpumask-locks \ + --num-trace-entries=0 --huge-unlink -r "$RPC_SOCKET" >"$log" 2>&1 & + # Record ownership before RPC setup, so failed startup can also be cleaned up. + TARGET_PID=$! + until nof_target_rpc bdev_get_bdevs >"${log%.log}-rpc-ready.log" 2>&1; do + kill -0 "$TARGET_PID" 2>/dev/null && (( SECONDS < deadline )) || return 1 + sleep 0.1 + done + nof_target_rpc nvmf_create_transport -t TCP >"$NOF_TARGET_SETUP_LOG" || return + nof_target_rpc nvmf_create_subsystem "$TARGET_NQN" -a -s MC_NOF_E2E \ + >>"$NOF_TARGET_SETUP_LOG" +} + +nof_target_add_malloc_ns() { + local nsid=$1 size_mb=$2 block_size=$3 + nof_target_rpc bdev_malloc_create -b "Malloc$nsid" "$size_mb" "$block_size" \ + >>"$NOF_TARGET_SETUP_LOG" || return + nof_target_rpc nvmf_subsystem_add_ns "$TARGET_NQN" "Malloc$nsid" -n "$nsid" \ + >>"$NOF_TARGET_SETUP_LOG" +} + +nof_target_start_listener() { + nof_target_rpc nvmf_subsystem_add_listener "$TARGET_NQN" -t TCP \ + -a "$TARGET_HOST" -s "$TARGET_PORT" >>"$NOF_TARGET_SETUP_LOG" +} + +nof_target_stop() { + [[ -n "${TARGET_PID:-}" ]] || return 0 + # Also stop a target paused with SIGSTOP; only signal the runner's own child. + kill -KILL "$TARGET_PID" 2>/dev/null || true + wait "$TARGET_PID" 2>/dev/null || true + TARGET_PID="" + rm -f "$RPC_SOCKET" "$RPC_SOCKET.lock" +} diff --git a/mooncake-store/tests/e2e/readme.md b/mooncake-store/tests/e2e/readme.md index 4d3ac41aaf..1b049b0ae1 100644 --- a/mooncake-store/tests/e2e/readme.md +++ b/mooncake-store/tests/e2e/readme.md @@ -13,6 +13,7 @@ The E2E test suite includes several executable programs designed to test differe - **chaos_rand_test**: Long-term randomized chaos testing with configurable parameters. - **store_client_e2e.py**: Python `MooncakeDistributedStore` client that continuously issues `put/get` operations. - **run_nof_heartbeat_tcp_e2e.sh**: Scripted NoF heartbeat end-to-end test using a TCP SPDK target. +- **run_nof_controller_recovery_tcp_e2e.sh**: Verifies controller reset and I/O qpair reconnection across two workers after a TCP target restart. ## Parameters @@ -123,6 +124,58 @@ PRE_FAULT_SUCCESS_TARGET=10 BUILD_DIR=/path/to/build ./run_nof_heartbeat_tcp_e2e - In `CLIENT_GLOBAL_SEGMENT_SIZE=0` mode, the script verifies **NoF-only failure behavior** by checking that post-fault I/O starts failing after the NoF segment is removed. - Logs are written under `LOG_DIR` (default `/tmp/mooncake_nof_heartbeat_e2e`) and the final pass/fail summary is printed from `summary.log`. +### run_nof_controller_recovery_tcp_e2e.sh + +Runs one Store NoF worker pool with two workers and two namespaces on the same +controller. After verifying initial writes and reads, the runner pauses the +target, waits until both workers have pending I/O, kills the target and checks +that the requests fail within a timeout. It then recreates the target at the same +endpoint. The original pool and segment handles must recover and successfully +write/read a new pattern in both namespaces. Malloc bdev contents are not +expected to survive the target restart. + +Build with `USE_NOF=ON` and `BUILD_UNIT_TESTS=ON`, using the normal SPDK build +dependencies. Configure sufficient hugepages and access permissions for both the +client and target before running: + +```bash +cmake --build build --target nof_controller_recovery_e2e nof_worker_pool_config_test +BUILD_DIR="$PWD/build" SPDK_DIR="$PWD/extern/spdk" \ + ./mooncake-store/tests/e2e/run_nof_controller_recovery_tcp_e2e.sh +``` + +The script follows the other e2e runners' `BUILD_DIR`/`LOG_DIR` conventions. It +requires Linux, Python 3, GNU `timeout`, and a built SPDK `nvmf_tgt` plus `rpc.py`. +`LOG_DIR` must be empty; by default a new directory is created under `/tmp`. +`TARGET_PORT`, `TARGET_CORE_MASK` and `TIMEOUT_SEC` can be overridden. Only the +target and client started by this script are stopped on exit. + +The client uses the production SPDK initialization path. The target uses malloc +bdevs backed by hugepages, with PCI disabled and hugepage backing files unlinked +at initialization so they do not survive a forced stop. The script does not +configure the host's hugepage pool or mount hugetlbfs. This is a manually invoked +integration test, not an automatic CTest target. + +The target lifecycle is shared through `nof_target.sh`. Runners source this +file and provide `SPDK_DIR`, `RPC_SOCKET`, `TARGET_NQN`, `TARGET_HOST` and +`TARGET_PORT`; `TARGET_CORE_MASK`, `TARGET_MEM_MB` and `TIMEOUT_SEC` are optional. +For example, a runner can configure two different namespaces as follows: + +```bash +source "$SCRIPT_DIR/nof_target.sh" +nof_target_start "$LOG_DIR/target-1.log" +nof_target_add_malloc_ns 1 32 512 # NSID, capacity in MiB, block size in bytes +nof_target_add_malloc_ns 2 32 4096 +nof_target_start_listener +``` + +Call `nof_target_add_malloc_ns` once per namespace. IDs are explicit so restarting +the target can recreate the same layout. The helper retains `TARGET_PID` for +fault injection; `nof_target_stop` kills and reaps that target and removes its +RPC socket. Each runner owns its EXIT trap and calls this cleanup function, +including when startup fails. The helper does not configure hugepages or manage +client processes. + ### store_client_e2e.py **Brief**: A standalone Python workload generator built on `MooncakeDistributedStore`. It continuously issues `put/get` against the configured master/metadata pair and prints `put_ok/get_ok/put_fail/get_fail` lines that can be consumed by shell scripts. diff --git a/mooncake-store/tests/e2e/run_nof_controller_recovery_tcp_e2e.sh b/mooncake-store/tests/e2e/run_nof_controller_recovery_tcp_e2e.sh new file mode 100755 index 0000000000..c4e3a1b86c --- /dev/null +++ b/mooncake-store/tests/e2e/run_nof_controller_recovery_tcp_e2e.sh @@ -0,0 +1,99 @@ +#!/usr/bin/env bash +# Two workers / namespaces recover through a real TCP malloc-bdev target restart. +set -euo pipefail + +SCRIPT_DIR=$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd) +REPO_ROOT=$(cd -- "$SCRIPT_DIR/../../.." && pwd) +BUILD_DIR=${BUILD_DIR:-"$REPO_ROOT/build"} +SPDK_DIR=${SPDK_DIR:-"$REPO_ROOT/extern/spdk"} +LOG_DIR=${LOG_DIR:-$(mktemp -d /tmp/mooncake_nof_recovery_e2e.XXXXXX)} +TIMEOUT_SEC=${TIMEOUT_SEC:-30} +TARGET_HOST=127.0.0.1 +TARGET_PORT=${TARGET_PORT:-$(python3 - <<'PY' +import socket +with socket.socket() as sock: + sock.bind(('127.0.0.1', 0)) + print(sock.getsockname()[1]) +PY +)} +TARGET_NQN=${TARGET_NQN:-nqn.2016-06.io.spdk:mooncake-recovery-$$} +TARGET_CORE_MASK=${TARGET_CORE_MASK:-0x1} +TARGET_BIN="$SPDK_DIR/build/bin/nvmf_tgt" +CLIENT_BIN="$BUILD_DIR/mooncake-store/tests/e2e/nof_controller_recovery_e2e" +TARGET_PID="" +CLIENT_PID="" +source "$SCRIPT_DIR/nof_target.sh" + +cleanup() { + local rc=$? + trap - EXIT + if [[ -n "$CLIENT_PID" ]]; then + # timeout creates a process group containing the client. + kill -KILL -- "-$CLIENT_PID" 2>/dev/null || true + wait "$CLIENT_PID" 2>/dev/null || true + fi + nof_target_stop + if (( rc != 0 )); then + echo "FAILED: NoF controller recovery (logs: $LOG_DIR)" >&2 + tail -n 60 "$LOG_DIR/client.log" "$LOG_DIR/target-1.log" \ + "$LOG_DIR/target-2.log" 2>/dev/null >&2 || true + fi + exit "$rc" +} + +[[ "$TIMEOUT_SEC" =~ ^[1-9][0-9]*$ ]] || { echo "Invalid TIMEOUT_SEC" >&2; exit 2; } +[[ -x "$CLIENT_BIN" && -x "$TARGET_BIN" ]] || { + echo "Build nof_controller_recovery_e2e (USE_NOF=ON) and SPDK nvmf_tgt first" >&2 + exit 2 +} +# Keep previous results; never remove an existing run's directory. +mkdir -p "$LOG_DIR" +[[ -z "$(ls -A "$LOG_DIR")" ]] || { echo "LOG_DIR must be empty" >&2; exit 2; } +LOG_DIR=$(cd -- "$LOG_DIR" && pwd) +RPC_SOCKET="$LOG_DIR/spdk.sock" +[[ ${#RPC_SOCKET} -lt 104 ]] || { echo "LOG_DIR path is too long for an RPC socket" >&2; exit 2; } +trap cleanup EXIT +export LD_LIBRARY_PATH="$BUILD_DIR/mooncake-common${LD_LIBRARY_PATH:+:$LD_LIBRARY_PATH}" + +start_target() { + nof_target_start "$LOG_DIR/target-$1.log" || return + nof_target_add_malloc_ns 1 32 512 || return + nof_target_add_malloc_ns 2 32 4096 || return + nof_target_start_listener +} + +wait_for_pattern() { + local pattern=$1 deadline=$((SECONDS + TIMEOUT_SEC)) + until grep -q "$pattern" "$LOG_DIR/client.log"; do + kill -0 "$CLIENT_PID" 2>/dev/null && (( SECONDS < deadline )) || return 1 + sleep 0.1 + done +} + +start_target 1 +ENDPOINT="trtype:TCP adrfam:IPv4 traddr:$TARGET_HOST trsvcid:$TARGET_PORT subnqn:$TARGET_NQN" +MC_NVME_FABRICS_CONNECT_TIMEOUT_US=500000 MC_NOF_SUBMIT_CHUNK_BYTES=65536 \ +MC_NOF_INFLIGHT_BYTES_LIMIT=131072 \ + timeout --kill-after=5 "$((TIMEOUT_SEC * 6))" "$CLIENT_BIN" \ + --endpoint="$ENDPOINT" --control_dir="$LOG_DIR" --timeout_sec="$TIMEOUT_SEC" \ + >"$LOG_DIR/client.log" 2>&1 & +CLIENT_PID=$! +wait_for_pattern '^PASS initial_io$' + +kill -STOP "$TARGET_PID" +deadline=$((SECONDS + TIMEOUT_SEC)) +until grep -q '^State:.*T' "/proc/$TARGET_PID/status"; do + (( SECONDS < deadline )) || exit 1 + sleep 0.01 +done +touch "$LOG_DIR/target-paused" +wait_for_pattern '^READY fault_io_pending$' +nof_target_stop +wait_for_pattern '^PASS failed_io_drained$' + +start_target 2 +touch "$LOG_DIR/target-restarted" +wait_for_pattern '^PASS controller_recovery$' +wait "$CLIENT_PID" +CLIENT_PID="" +echo "PASSED: two-worker NoF controller recovery (logs: $LOG_DIR)" diff --git a/mooncake-store/tests/nof_worker_pool_config_test.cpp b/mooncake-store/tests/nof_worker_pool_config_test.cpp index 0a56d0fb07..0eca533c61 100644 --- a/mooncake-store/tests/nof_worker_pool_config_test.cpp +++ b/mooncake-store/tests/nof_worker_pool_config_test.cpp @@ -97,12 +97,12 @@ TEST_F(NoFWorkerPoolConfigTest, PoolConstructionUsesCachedWorkerCount) { std::filesystem::directory_iterator("/proc/self/task"), std::filesystem::directory_iterator()); }; - const auto before = count_threads(); SpdkNofWorkerPool first; - EXPECT_EQ(count_threads() - before, 2); + // Include the shared controller thread in the baseline. + const auto before = count_threads(); ASSERT_EQ(setenv("MC_NOF_WORKERS", "3", 1), 0); SpdkNofWorkerPool second; - EXPECT_EQ(count_threads() - before, 4); + EXPECT_EQ(count_threads() - before, 2); EXPECT_EQ(NoFWorkerPoolConfig::AtFirstUse().worker_count, 2); } #endif From d543d48eba44f847a8b04a8176714f94332be05f Mon Sep 17 00:00:00 2001 From: Cage Chen Date: Tue, 20 Oct 2026 16:23:22 +0800 Subject: [PATCH 4/4] [Bugfix][Store] Recover master NoF heartbeat connections Master heartbeat probes keep reusing cached controllers and qpairs after connection failures. Subsequent probes can keep failing after the target recovers, eventually causing healthy segments to be removed. Drain all probe qpairs on the affected controller before disconnecting it. Reconnect on the next heartbeat, refresh namespace handles, and recreate qpairs as needed. Use the existing heartbeat thread to advance recovery and finish outstanding I/O safely before reusing probe buffers. Signed-off-by: Cage Chen --- mooncake-store/include/spdk/spdk_wrapper.h | 13 +- mooncake-store/src/master_service.cpp | 4 + mooncake-store/src/spdk/spdk_wrapper.cpp | 284 +++++++++++++++--- mooncake-store/src/transfer_task.cpp | 8 +- mooncake-store/tests/e2e/CMakeLists.txt | 6 + .../tests/e2e/nof_heartbeat_recovery_e2e.cpp | 190 ++++++++++++ mooncake-store/tests/e2e/readme.md | 17 ++ .../e2e/run_nof_heartbeat_recovery_tcp_e2e.sh | 153 ++++++++++ 8 files changed, 624 insertions(+), 51 deletions(-) create mode 100644 mooncake-store/tests/e2e/nof_heartbeat_recovery_e2e.cpp create mode 100755 mooncake-store/tests/e2e/run_nof_heartbeat_recovery_tcp_e2e.sh diff --git a/mooncake-store/include/spdk/spdk_wrapper.h b/mooncake-store/include/spdk/spdk_wrapper.h index 2c69fb78f8..22b6bf46e2 100644 --- a/mooncake-store/include/spdk/spdk_wrapper.h +++ b/mooncake-store/include/spdk/spdk_wrapper.h @@ -46,10 +46,10 @@ class SpdkWrapper { int64_t NvmePollProcessCompletion(nof_seg_handle *seg, uint32_t complete_per_seg); - // When recovery_enabled is true, only prepare the segment metadata; + // When client_managed is true, only prepare the segment metadata; // the owner worker creates the qpair. nof_seg_handle *OpenNofSegment(const std::string &tr_str, - bool recovery_enabled = false); + bool client_managed = false); uint32_t GetBlockSize(const nof_seg_handle *seg_handle); @@ -60,6 +60,9 @@ class SpdkWrapper { bool ProbeNofSegment(const std::string &tr_str, uint32_t timeout_ms, std::string *error_reason = nullptr); + void PollProbeControllers(); + void ShutdownProbeConnections(); + /** @brief Register external memory with SPDK for NoF zero-copy transfers. * * Memory returned by Alloc() is already registered with SPDK; use this to @@ -127,8 +130,8 @@ class SpdkWrapper { void OwnerDrained(ctrlr_info *ctrlr, WorkerId owner); void UnregisterOwner(ctrlr_info *ctrlr, WorkerId owner); - int SetupClientQpairIfNeeded(nof_seg_handle *seg); - bool DrainClientQpair(nof_seg_handle *seg); + int SetupQpairIfNeeded(nof_seg_handle *seg); + bool DrainQpair(nof_seg_handle *seg); private: struct ProbeBuffer { @@ -172,6 +175,8 @@ class SpdkWrapper { void ReplenishProbeRequestContextPoolLocked(size_t count); static void ProbeReadComplete(void *ctx, const struct spdk_nvme_cpl *cpl); + void MarkProbeControllerDraining(ctrlr_info *info); + void ControllerLoop(); std::atomic initialized{false}; diff --git a/mooncake-store/src/master_service.cpp b/mooncake-store/src/master_service.cpp index 053e75f59c..5f052a2b5d 100644 --- a/mooncake-store/src/master_service.cpp +++ b/mooncake-store/src/master_service.cpp @@ -12411,9 +12411,11 @@ bool MasterService::TryUnmountNoFSegmentByHeartbeat( return true; } +#ifdef USE_NOF void MasterService::NofHeartbeatThreadFunc() { size_t next_probe_index = 0; while (nof_heartbeat_running_) { + SpdkWrapper::GetInstance().PollProbeControllers(); auto now = std::chrono::steady_clock::now(); std::vector mounted_segments; nof_segment_manager_.GetMountedSegmentsSnapshot(mounted_segments); @@ -12561,7 +12563,9 @@ void MasterService::NofHeartbeatThreadFunc() { TryUnmountNoFSegmentByHeartbeat(*probe_target, error_reason); } } + SpdkWrapper::GetInstance().ShutdownProbeConnections(); } +#endif tl::expected MasterService::ValidateDynamicReplicaPendingForCopyStart( diff --git a/mooncake-store/src/spdk/spdk_wrapper.cpp b/mooncake-store/src/spdk/spdk_wrapper.cpp index 76c0043b68..00b5bd739d 100644 --- a/mooncake-store/src/spdk/spdk_wrapper.cpp +++ b/mooncake-store/src/spdk/spdk_wrapper.cpp @@ -80,12 +80,22 @@ struct tr_info { }; struct ctrlr_info { - enum class State { Ready, Draining, Disconnecting, Reconnecting, Backoff }; + enum class State { + Ready, + Draining, + Disconnecting, + Reconnecting, + Backoff, + // Probe-only: wait for the next heartbeat to initiate reconnection. + // Clients transition directly from Disconnecting to Reconnecting. + Disconnected + }; struct spdk_nvme_ctrlr *ctrlr{nullptr}; - // Fixed at creation, probe-only controllers leave recovery disabled. - bool recovery_enabled{false}; + // Fixed at creation: true for client workers and ControllerLoop; + // false for the Master heartbeat thread. + bool client_managed{false}; State state{State::Ready}; // One entry per owner worker, shared by its namespaces on this controller. std::set owners; @@ -398,7 +408,7 @@ int SpdkWrapper::ConnectController(const struct spdk_nvme_transport_id *trid, } nof_seg_handle *SpdkWrapper::OpenNofSegment(const std::string &tr_str, - bool recovery_enabled) { + bool client_managed) { // ConnectController -> spdk_nvme_probe() uses env-provided sockets and // memory, so the SPDK env must be up. Do not rely on an earlier // RegisterMemory() to have initialized it: callers now skip the @@ -424,10 +434,10 @@ nof_seg_handle *SpdkWrapper::OpenNofSegment(const std::string &tr_str, if (it == connected_ctrlrs.end()) { auto new_info = std::make_unique(); info = new_info.get(); - info->recovery_enabled = recovery_enabled; + info->client_managed = client_managed; ret = ConnectController(&tr.trid, info); - if (ret != 0) { + if (ret != 0 || info->ctrlr == nullptr) { return nullptr; } @@ -437,56 +447,56 @@ nof_seg_handle *SpdkWrapper::OpenNofSegment(const std::string &tr_str, } } - // SPDK may reuse a controller for the same transport ID. Never enable - // recovery around a legacy probe connection that has no worker owner. - if (info->recovery_enabled != recovery_enabled) { + // SPDK may reuse a controller for the same transport ID. Its ownership + // mode must remain fixed: client workers or the Master heartbeat thread. + if (info->client_managed != client_managed) { LOG(ERROR) << "Cannot mix probe and recoverable client ownership for " << tr.ctrlr_key; return nullptr; } nof_seg_handle *seg_handle = nullptr; - struct spdk_nvme_qpair *qpair = nullptr; - struct spdk_nvme_ns *ns = nullptr; { auto &ns_seg = info->ns_seg; std::lock_guard lock(info->ns_mutex); if (info->state != ctrlr_info::State::Ready) return nullptr; auto ns_it = ns_seg.find(tr.ns); - if (ns_it != ns_seg.end()) { + if (client_managed && ns_it != ns_seg.end()) { return ns_it->second.get(); } - if (spdk_nvme_ctrlr_is_active_ns(info->ctrlr, tr.ns)) { - ns = spdk_nvme_ctrlr_get_ns(info->ctrlr, tr.ns); - } else { + if (!spdk_nvme_ctrlr_is_active_ns(info->ctrlr, tr.ns)) { LOG(ERROR) << "spdk_nvme_ctrlr_is_active_ns failed"; return nullptr; } + auto *ns = spdk_nvme_ctrlr_get_ns(info->ctrlr, tr.ns); - auto new_seg = std::make_unique(); - if (!recovery_enabled) { - qpair = spdk_nvme_ctrlr_alloc_io_qpair(info->ctrlr, nullptr, 0); - if (!qpair) { - LOG(ERROR) << "alloc spdk_nvme_qpair failed"; - return nullptr; - } + if (ns_it == ns_seg.end()) { + ns_it = + ns_seg.emplace(tr.ns, std::make_unique()).first; } - new_seg->qpair = qpair; - new_seg->ns = ns; - new_seg->controller = info; - new_seg->block_size = spdk_nvme_ns_get_sector_size(ns); - seg_handle = new_seg.get(); - ns_seg[tr.ns] = std::move(new_seg); + seg_handle = ns_it->second.get(); + seg_handle->ns = ns; + seg_handle->controller = info; + seg_handle->block_size = spdk_nvme_ns_get_sector_size(ns); } + if (client_managed) return seg_handle; + + // The heartbeat owner also advances recovery, so no worker barrier is + // needed for probe qpairs. Do not hold ns_mutex across qpair creation. + ret = SetupQpairIfNeeded(seg_handle); + if (ret != 0) { + if (ret == -ENXIO) MarkProbeControllerDraining(info); + return nullptr; + } return seg_handle; } uint32_t SpdkWrapper::GetBlockSize(const nof_seg_handle *seg_handle) { if (!seg_handle) return INVALID_BLOCK_SIZE; - if (!seg_handle->controller->recovery_enabled) { + if (!seg_handle->controller->client_managed) { return seg_handle->ns ? spdk_nvme_ns_get_sector_size(seg_handle->ns) : INVALID_BLOCK_SIZE; } @@ -546,6 +556,8 @@ SpdkWrapper::ProbeBuffer *SpdkWrapper::GetOrCreateProbeBuffer( bool SpdkWrapper::ProbeNofSegment(const std::string &tr_str, uint32_t timeout_ms, std::string *error_reason) { + const auto deadline = std::chrono::steady_clock::now() + + std::chrono::milliseconds(timeout_ms); if (!InitializeEnv()) { if (error_reason) { *error_reason = "spdk_env_init_fail"; @@ -553,6 +565,53 @@ bool SpdkWrapper::ProbeNofSegment(const std::string &tr_str, return false; } + tr_info tr; + if (ParseTransPortStr(tr_str, &tr) != 0) { + if (error_reason) *error_reason = "invalid_endpoint"; + return false; + } + ctrlr_info *info = nullptr; + { + std::lock_guard lock(ctrlrs_mutex); + auto it = connected_ctrlrs.find(tr.ctrlr_key); + if (it != connected_ctrlrs.end()) info = it->second.get(); + } + if (info && info->client_managed) { + if (error_reason) *error_reason = "controller_owner_mismatch"; + return false; + } + + // Attempt at most one reconnect for this heartbeat. + using State = ctrlr_info::State; + bool attempted = info && info->state == State::Reconnecting; + while (info && info->state != State::Ready) { + if (std::chrono::steady_clock::now() >= deadline) { + if (error_reason) *error_reason = "recovery_in_progress"; + return false; + } + if (info->state == State::Disconnected && !attempted) { + { + std::lock_guard lock(info->ns_mutex); + info->state = State::Reconnecting; + } + spdk_nvme_ctrlr_reconnect_async(info->ctrlr); + attempted = true; + LOG(INFO) << "nof_probe_recovery reconnect ctrlr=" << info; + } + PollProbeControllers(); + if (attempted && info->state != State::Ready && + info->state != State::Reconnecting) { + if (error_reason) *error_reason = "reconnect_failed"; + return false; + } + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + } + + if (std::chrono::steady_clock::now() >= deadline) { + if (error_reason) *error_reason = "prepare_timeout"; + return false; + } + nof_seg_handle *seg_handle = OpenNofSegment(tr_str); if (!seg_handle) { if (error_reason) { @@ -560,9 +619,18 @@ bool SpdkWrapper::ProbeNofSegment(const std::string &tr_str, } return false; } + info = seg_handle->controller; + if (std::chrono::steady_clock::now() >= deadline) { + // Attach or qpair setup exhausted the shared budget. Drain any pending + // qpair connection instead of leaving it unpolled after returning. + MarkProbeControllerDraining(info); + if (error_reason) *error_reason = "open_timeout"; + return false; + } uint32_t block_size = GetBlockSize(seg_handle); if (block_size == INVALID_BLOCK_SIZE || block_size == 0) { + MarkProbeControllerDraining(info); if (error_reason) { *error_reason = "invalid_block_size"; } @@ -572,32 +640,57 @@ bool SpdkWrapper::ProbeNofSegment(const std::string &tr_str, ProbeBuffer *probe_buffer = GetOrCreateProbeBuffer(tr_str, block_size, error_reason); if (!probe_buffer || !probe_buffer->ptr) { + MarkProbeControllerDraining(info); return false; } ProbeRequestContext *probe_ctx = AcquireProbeRequestContext(); + if (std::chrono::steady_clock::now() >= deadline) { + // Buffer/context preparation used the remaining budget. No read owns + // this context yet; a pending qpair connection still needs cleanup. + RecycleProbeRequestContext(probe_ctx); + MarkProbeControllerDraining(info); + if (error_reason) *error_reason = "prepare_timeout"; + return false; + } int ret = SubmitRequest(seg_handle, probe_buffer->ptr, 0, 1, kSpdkNofOpRead, ProbeReadComplete, probe_ctx); if (ret != 0) { RecycleProbeRequestContext(probe_ctx); + MarkProbeControllerDraining(info); if (error_reason) { *error_reason = "submit_fail"; } return false; } - auto deadline = std::chrono::steady_clock::now() + - std::chrono::milliseconds(timeout_ms); + bool connection_error = false; while (!probe_ctx->done.load(std::memory_order_acquire) && std::chrono::steady_clock::now() < deadline) { - NvmePollProcessCompletion(seg_handle, 0); - std::this_thread::sleep_for(std::chrono::milliseconds(1)); + // Process admin events and advance recovery while waiting for the read. + // Check controller state before polling I/O. + PollProbeControllers(); + + if (info->state != State::Ready || + NvmePollProcessCompletion(seg_handle, 0) < 0 || + spdk_nvme_qpair_get_failure_reason(seg_handle->qpair) != + SPDK_NVME_QPAIR_FAILURE_NONE) { + connection_error = true; + break; + } + if (!probe_ctx->done.load(std::memory_order_acquire)) { + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + } } - bool ok = probe_ctx->done.load(std::memory_order_acquire) && + bool done = probe_ctx->done.load(std::memory_order_acquire); + bool timed_out = std::chrono::steady_clock::now() >= deadline; + bool ok = !connection_error && !timed_out && done && probe_ctx->success.load(std::memory_order_acquire); if (!ok && error_reason) { - if (!probe_ctx->done.load(std::memory_order_acquire)) { + if (connection_error) { + *error_reason = "connection_error"; + } else if (!done || timed_out) { *error_reason = "completion_timeout"; } else { std::lock_guard lock(probe_ctx->error_mutex); @@ -607,9 +700,117 @@ bool SpdkWrapper::ProbeNofSegment(const std::string &tr_str, } } + // A timeout does not imply a broken connection. Drain unfinished reads + // before reusing their buffers; completion/abort callbacks recycle the + // request contexts. + if (connection_error || !done) MarkProbeControllerDraining(info); + return ok; } +void SpdkWrapper::MarkProbeControllerDraining(ctrlr_info *info) { + assert(!info->client_managed); + std::lock_guard lock(info->ns_mutex); + if (info->state != ctrlr_info::State::Ready) return; + info->state = ctrlr_info::State::Draining; + LOG(WARNING) << "nof_probe_recovery begin ctrlr=" << info + << " namespaces=" << info->ns_seg.size(); +} + +void SpdkWrapper::PollProbeControllers() { + using State = ctrlr_info::State; + std::vector controllers; + { + std::lock_guard lock(ctrlrs_mutex); + for (auto &[_, info] : connected_ctrlrs) { + if (!info->client_managed) controllers.push_back(info.get()); + } + } + for (auto *info : controllers) { + switch (info->state) { + case State::Ready: { + int rc = spdk_nvme_ctrlr_process_admin_completions(info->ctrlr); + if (rc >= 0) break; + MarkProbeControllerDraining(info); + } + [[fallthrough]]; + case State::Draining: { + // Give every qpair a chance to finish teardown, even if an + // earlier one is still pending. Disconnect the controller + // only after all qpairs and their outstanding I/O are drained. + bool drained = true; + for (auto &[_, seg] : info->ns_seg) { + if (!DrainQpair(seg.get())) drained = false; + } + if (!drained) break; + // Reset may invalidate SPDK namespace pointers. Keep the + // handles; OpenNofSegment refreshes namespaces on demand. + for (auto &[_, seg] : info->ns_seg) seg->ns = nullptr; + int rc = spdk_nvme_ctrlr_disconnect(info->ctrlr); + std::lock_guard lock(info->ns_mutex); + info->state = + rc == -ENXIO ? State::Disconnected : State::Disconnecting; + if (info->state == State::Disconnected) break; + } + [[fallthrough]]; + case State::Disconnecting: + if (spdk_nvme_ctrlr_process_admin_completions(info->ctrlr) == + -ENXIO) { + std::lock_guard lock(info->ns_mutex); + info->state = State::Disconnected; + LOG(INFO) + << "nof_probe_recovery disconnected ctrlr=" << info; + } + break; + case State::Disconnected: + // Cleanup is complete. Idle polling must not start retries; + // the next heartbeat probe decides whether to reconnect. + break; + case State::Reconnecting: { + int rc = spdk_nvme_ctrlr_reconnect_poll_async(info->ctrlr); + if (rc == -EAGAIN) break; + std::lock_guard lock(info->ns_mutex); + if (rc != 0) { + // Clean up the failed attempt on the next poll. A new + // reconnect still requires another heartbeat probe. + info->state = State::Draining; + LOG(WARNING) << "nof_probe_recovery failed ctrlr=" << info + << " rc=" << rc; + } else { + info->state = State::Ready; + LOG(INFO) << "nof_probe_recovery ready ctrlr=" << info; + } + break; + } + case State::Backoff: + assert(false && "Backoff is only used by client recovery"); + break; + } + } +} + +void SpdkWrapper::ShutdownProbeConnections() { + std::vector controllers; + { + std::lock_guard lock(ctrlrs_mutex); + for (auto &[_, info] : connected_ctrlrs) { + if (!info->client_managed) controllers.push_back(info.get()); + } + } + // A reconnect holds SPDK's controller mutex across polling calls. Finish + // it on this owner even on shutdown; then drain without starting retries. + while (true) { + bool drained = true; + for (auto *info : controllers) { + MarkProbeControllerDraining(info); + if (info->state != ctrlr_info::State::Disconnected) drained = false; + } + if (drained) break; + PollProbeControllers(); + std::this_thread::sleep_for(std::chrono::milliseconds(100)); + } +} + void SpdkWrapper::RegisterRecoveryTaskPoster(RecoveryTaskPoster poster) { // Serialize registration from concurrently constructed worker pools. std::lock_guard lock(controller_mutex_); @@ -630,7 +831,7 @@ bool SpdkWrapper::BindSegmentToWorker(nof_seg_handle *seg, WorkerId owner, auto *info = seg->controller; std::lock_guard lock(info->ns_mutex); - if (!info->recovery_enabled || info->state != ctrlr_info::State::Ready || + if (!info->client_managed || info->state != ctrlr_info::State::Ready || !seg->ns || seg->owner.first) return false; @@ -648,7 +849,7 @@ bool SpdkWrapper::BindSegmentToWorker(nof_seg_handle *seg, WorkerId owner, void SpdkWrapper::RequestRecovery(ctrlr_info *info) { std::lock_guard lock(info->ns_mutex); - if (!info->recovery_enabled || info->state != ctrlr_info::State::Ready) + if (!info->client_managed || info->state != ctrlr_info::State::Ready) return; info->state = ctrlr_info::State::Draining; @@ -694,7 +895,7 @@ void SpdkWrapper::UnregisterOwner(ctrlr_info *info, WorkerId owner) { controller_cv_.notify_one(); } -int SpdkWrapper::SetupClientQpairIfNeeded(nof_seg_handle *seg) { +int SpdkWrapper::SetupQpairIfNeeded(nof_seg_handle *seg) { // Called only on its owner. The common case never takes a controller lock. if (seg->qpair) return 0; @@ -705,9 +906,6 @@ int SpdkWrapper::SetupClientQpairIfNeeded(nof_seg_handle *seg) { if (!seg->ns) return -ENODEV; } - // This registered owner cannot acknowledge Cleanup while creating here. - // The cleanup barrier protects the controller without holding admission - // locked across a potentially slow transport connect. spdk_nvme_io_qpair_opts opts; spdk_nvme_ctrlr_get_default_io_qpair_opts(info->ctrlr, &opts, sizeof(opts)); opts.async_mode = true; @@ -726,7 +924,7 @@ int SpdkWrapper::SetupClientQpairIfNeeded(nof_seg_handle *seg) { return rc == 0 ? 0 : -ENXIO; } -bool SpdkWrapper::DrainClientQpair(nof_seg_handle *seg) { +bool SpdkWrapper::DrainQpair(nof_seg_handle *seg) { if (!seg->qpair) return true; if (!seg->disconnecting) { @@ -771,7 +969,7 @@ void SpdkWrapper::ControllerLoop() { { std::lock_guard lock(ctrlrs_mutex); for (auto &[_, info] : connected_ctrlrs) { - if (info->recovery_enabled) controllers.push_back(info.get()); + if (info->client_managed) controllers.push_back(info.get()); } } diff --git a/mooncake-store/src/transfer_task.cpp b/mooncake-store/src/transfer_task.cpp index 8e18bb9170..0c360583ef 100644 --- a/mooncake-store/src/transfer_task.cpp +++ b/mooncake-store/src/transfer_task.cpp @@ -525,7 +525,7 @@ struct SpdkNofWorkerPool::WorkerContext { bool drained = true; for (auto* seg : state.draining_segments) - drained &= wrapper.DrainClientQpair(seg); + drained &= wrapper.DrainQpair(seg); if (!drained) continue; for (auto& [_, qos] : seg_to_qos) { @@ -629,7 +629,7 @@ void SpdkNofWorkerPool::workerThread(int work_idx) { if (nof_qos->controller_state->phase != Phase::Running || nof_qos->Empty()) continue; - int connect_rc = wrapper.SetupClientQpairIfNeeded(seg_handle); + int connect_rc = wrapper.SetupQpairIfNeeded(seg_handle); if (connect_rc != 0) { if (connect_rc == -ENXIO) wrapper.RequestRecovery(wrapper.GetController(seg_handle)); @@ -791,12 +791,12 @@ nof_seg_handle* SpdkNofWorkerPool::OpenNofSegment(const std::string& tr_str) { // the local registration above so it finds this // controller; otherwise the controller could retain an // owner referencing a destroyed pool. Skip only - // SpdkWrapper::SetupClientQpairIfNeeded() below. + // SpdkWrapper::SetupQpairIfNeeded() below. if (ctx.pool.shutdown_.load() || state.phase != WorkerContext::Phase::Running) return; - int rc = ctx.wrapper.SetupClientQpairIfNeeded(seg); + int rc = ctx.wrapper.SetupQpairIfNeeded(seg); if (rc == -ENXIO) ctx.wrapper.RequestRecovery(ctrlr); if (rc != 0 && rc != -EAGAIN) { LOG(WARNING) diff --git a/mooncake-store/tests/e2e/CMakeLists.txt b/mooncake-store/tests/e2e/CMakeLists.txt index bb3f50faa3..b895f6ad77 100644 --- a/mooncake-store/tests/e2e/CMakeLists.txt +++ b/mooncake-store/tests/e2e/CMakeLists.txt @@ -105,4 +105,10 @@ if(USE_NOF) target_link_libraries( nof_controller_recovery_e2e PRIVATE mooncake_store gflags::gflags glog::glog pthread) + + # Run through run_nof_heartbeat_recovery_tcp_e2e.sh. + add_executable(nof_heartbeat_recovery_e2e nof_heartbeat_recovery_e2e.cpp) + target_link_libraries( + nof_heartbeat_recovery_e2e PRIVATE mooncake_store gflags::gflags + glog::glog pthread) endif() diff --git a/mooncake-store/tests/e2e/nof_heartbeat_recovery_e2e.cpp b/mooncake-store/tests/e2e/nof_heartbeat_recovery_e2e.cpp new file mode 100644 index 0000000000..ff18de581a --- /dev/null +++ b/mooncake-store/tests/e2e/nof_heartbeat_recovery_e2e.cpp @@ -0,0 +1,190 @@ +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "master_client.h" + +DEFINE_string(master_server, "", "Address of the runner's mooncake_master"); +DEFINE_string(endpoint, "", "TCP controller endpoint without a namespace ID"); +DEFINE_string(control_dir, "", + "Target lifecycle markers written by the runner"); +DEFINE_uint32(timeout_sec, 45, "Timeout for each test phase"); + +namespace { +using namespace mooncake; +using namespace std::chrono_literals; +using Clock = std::chrono::steady_clock; +constexpr size_t kNamespaceSize = 32 * 1024 * 1024; + +void Require(bool ok, const std::string& message) { + if (!ok) throw std::runtime_error(message); +} + +template +void WaitFor(Predicate ready, const std::string& message) { + const auto deadline = + Clock::now() + std::chrono::seconds(FLAGS_timeout_sec); + while (!ready()) { + Require(Clock::now() < deadline, message); + std::this_thread::sleep_for(100ms); + } +} + +void WaitForMarker(const char* name) { + const auto path = std::filesystem::path(FLAGS_control_dir) / name; + WaitFor([&] { return std::filesystem::exists(path); }, + "Timed out waiting for " + path.string()); +} + +struct Observations { + size_t successes{0}; + size_t failures{0}; + size_t timeouts{0}; + size_t unmounts{0}; +}; + +using Segments = std::array; + +std::array Observe(const Segments& segments) { + std::ifstream log(std::filesystem::path(FLAGS_control_dir) / "master.log"); + Require(log.is_open(), "Cannot read master.log"); + std::array observations{}; + std::string line; + while (std::getline(log, line)) { + for (size_t i = 0; i < segments.size(); ++i) { + if (line.find(", segment_name=" + segments[i].name + ",") == + std::string::npos) + continue; + auto& observation = observations[i]; + if (line.find(", action=nof_heartbeat_success,") != + std::string::npos) { + ++observation.successes; + } else if (line.find(", action=nof_heartbeat_failure,") != + std::string::npos) { + ++observation.failures; + if (line.find("reason=completion_timeout") != std::string::npos) + ++observation.timeouts; + } else if (line.find( + ", action=unmount_nof_segment_by_heartbeat,") != + std::string::npos) { + ++observation.unmounts; + } + } + } + return observations; +} + +void CheckRegistrations(MasterClient& client, const Segments& expected) { + const auto mounted = client.GetAllNoFSegments(); + Require(mounted.has_value(), "GetAllNoFSegments RPC failed"); + Require(mounted->size() == expected.size(), + "Mounted namespace count changed during recovery"); + for (const auto& segment : expected) + Require(std::any_of(mounted->begin(), mounted->end(), + [&](const NoFSegment& current) { + return current.id == segment.id && + current.name == segment.name && + current.te_endpoint == + segment.te_endpoint && + current.size == segment.size && + current.base == segment.base; + }), + "Recovery lost the original registration: " + segment.name); +} + +void Run() { + Require(!FLAGS_master_server.empty() && !FLAGS_endpoint.empty() && + !FLAGS_control_dir.empty() && FLAGS_timeout_sec > 30, + "Specify --master_server, --endpoint, --control_dir and " + "--timeout_sec > 30"); + MasterClient client(generate_uuid()); + WaitFor( + [&] { return client.Connect(FLAGS_master_server) == ErrorCode::OK; }, + "Master RPC did not become ready"); + Segments segments; + for (size_t i = 0; i < segments.size(); ++i) { + auto& segment = segments[i]; + segment.id = generate_uuid(); + segment.name = "heartbeat_recovery_ns" + std::to_string(i + 1); + segment.base = 0; + segment.size = kNamespaceSize; + segment.te_endpoint = FLAGS_endpoint + " ns:" + std::to_string(i + 1); + Require(client.MountNoFSegment(segment).has_value(), + "MountNoFSegment RPC failed: " + segment.name); + } + CheckRegistrations(client, segments); + WaitFor( + [&] { + const auto observations = Observe(segments); + return observations[0].successes > 0 && + observations[1].successes > 0; + }, + "Initial heartbeats did not succeed on both namespaces"); + const auto initial = Observe(segments); + std::cout << "PASS initial_heartbeats" << std::endl; + + // SIGSTOP preserves TCP connections while preventing reads from completing. + WaitForMarker("target-paused"); + WaitFor( + [&] { + const auto observations = Observe(segments); + return observations[0].failures > initial[0].failures && + observations[1].failures > initial[1].failures && + observations[0].timeouts + observations[1].timeouts > + initial[0].timeouts + initial[1].timeouts; + }, + "Paused target did not produce a read timeout and failed probes"); + CheckRegistrations(client, segments); + const auto before_restart = Observe(segments); + std::cout << "PASS probe_timeout" << std::endl; + + WaitForMarker("target-restarted"); + WaitFor( + [&] { + const auto observations = Observe(segments); + return observations[0].successes > before_restart[0].successes && + observations[1].successes > before_restart[1].successes; + }, + "Original registrations did not recover successful heartbeats"); + CheckRegistrations(client, segments); + std::cout << "PASS heartbeats_recovered" << std::endl; + + // Persistent failure must still remove both original registrations. + WaitForMarker("target-down"); + WaitFor( + [&] { + const auto observations = Observe(segments); + return observations[0].unmounts > 0 && observations[1].unmounts > 0; + }, + "Namespaces were not removed by heartbeat"); + const auto mounted = client.GetAllNoFSegments(); + Require(mounted.has_value(), "GetAllNoFSegments RPC failed"); + Require(mounted->empty(), "Removed namespaces are still registered"); + std::cout << "PASS failed_namespaces_removed" << std::endl; + + // The runner checks SIGTERM and exit of the independent Master process. +} +} // namespace + +int main(int argc, char** argv) { + gflags::ParseCommandLineFlags(&argc, &argv, true); + google::InitGoogleLogging(argv[0]); + FLAGS_logtostderr = true; + try { + Run(); + std::cout << "PASS heartbeat_checks" << std::endl; + return 0; + } catch (const std::exception& error) { + std::cerr << "FAIL heartbeat_recovery: " << error.what() << std::endl; + return 1; + } +} diff --git a/mooncake-store/tests/e2e/readme.md b/mooncake-store/tests/e2e/readme.md index d82116fc48..416e5d986a 100644 --- a/mooncake-store/tests/e2e/readme.md +++ b/mooncake-store/tests/e2e/readme.md @@ -14,6 +14,7 @@ The E2E test suite includes several executable programs designed to test differe - **store_client_e2e.py**: Python `MooncakeDistributedStore` client that continuously issues `put/get` operations. - **run_nof_heartbeat_tcp_e2e.sh**: Scripted NoF heartbeat end-to-end test using a TCP SPDK target. - **run_nof_controller_recovery_tcp_e2e.sh**: Verifies controller reset and I/O qpair reconnection across two workers after a TCP target restart. +- **run_nof_heartbeat_recovery_tcp_e2e.sh**: Verifies Master heartbeat recovery for two namespaces after a probe timeout and TCP target restart. ## Parameters @@ -124,6 +125,22 @@ PRE_FAULT_SUCCESS_TARGET=10 BUILD_DIR=/path/to/build ./run_nof_heartbeat_tcp_e2e - In `CLIENT_GLOBAL_SEGMENT_SIZE=0` mode, the script verifies **NoF-only failure behavior** by checking that post-fault I/O starts failing after the NoF segment is removed. - Logs are written under `LOG_DIR` (default `/tmp/mooncake_nof_heartbeat_e2e`) and the final pass/fail summary is printed from `summary.log`. +### run_nof_heartbeat_recovery_tcp_e2e.sh + +Verifies recovery of two namespaces after a TCP target timeout/restart and +removal during a persistent outage, using an independent Master process. + +Requires `USE_NOF=ON`, `BUILD_UNIT_TESTS=ON`, a built SPDK `nvmf_tgt`, +and configured hugepages/access permissions. Run manually: + +```bash +cmake --build build --target mooncake_master nof_heartbeat_recovery_e2e +BUILD_DIR="$PWD/build" SPDK_DIR="$PWD/extern/spdk" \ + ./mooncake-store/tests/e2e/run_nof_heartbeat_recovery_tcp_e2e.sh +``` + +Logs are saved in a fresh `/tmp/mooncake_heartbeat_recovery.*` directory. + ### run_nof_controller_recovery_tcp_e2e.sh Runs one Store NoF worker pool with two workers and two namespaces on the same diff --git a/mooncake-store/tests/e2e/run_nof_heartbeat_recovery_tcp_e2e.sh b/mooncake-store/tests/e2e/run_nof_heartbeat_recovery_tcp_e2e.sh new file mode 100755 index 0000000000..e923d9c1a7 --- /dev/null +++ b/mooncake-store/tests/e2e/run_nof_heartbeat_recovery_tcp_e2e.sh @@ -0,0 +1,153 @@ +#!/usr/bin/env bash +# Real Master heartbeats recover two namespaces without registering them again. +set -euo pipefail + +SCRIPT_DIR=$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd) +REPO_ROOT=$(cd -- "$SCRIPT_DIR/../../.." && pwd) +BUILD_DIR=${BUILD_DIR:-"$REPO_ROOT/build"} +SPDK_DIR=${SPDK_DIR:-"$REPO_ROOT/extern/spdk"} +LOG_DIR=${LOG_DIR:-$(mktemp -d /tmp/mooncake_heartbeat_recovery.XXXXXX)} +TIMEOUT_SEC=${TIMEOUT_SEC:-45} +TARGET_HOST=127.0.0.1 +read -r target_port master_port metrics_port < <(python3 - <<'PY' +import socket +sockets = [socket.socket() for _ in range(3)] +try: + for sock in sockets: + sock.bind(('127.0.0.1', 0)) + print(*(sock.getsockname()[1] for sock in sockets)) +finally: + for sock in sockets: + sock.close() +PY +) +TARGET_PORT=${TARGET_PORT:-$target_port} +MASTER_RPC=${MASTER_RPC:-127.0.0.1:$master_port} +MASTER_HOST=${MASTER_RPC%:*} +MASTER_PORT=${MASTER_RPC##*:} +METRICS_PORT=${METRICS_PORT:-$metrics_port} +TARGET_NQN=${TARGET_NQN:-nqn.2016-06.io.spdk:mooncake-heartbeat-$$} +TEST_BIN="$BUILD_DIR/mooncake-store/tests/e2e/nof_heartbeat_recovery_e2e" +MASTER_BIN="$BUILD_DIR/mooncake-store/src/mooncake_master" +TARGET_PID="" +MASTER_PID="" +TEST_PID="" +source "$SCRIPT_DIR/nof_target.sh" + +stop_master() { + local deadline=$((SECONDS + $1)) rc=0 + kill -TERM "$MASTER_PID" || return 1 + while kill -0 "$MASTER_PID" 2>/dev/null; do + (( SECONDS < deadline )) || return 1 + sleep 0.1 + done + wait "$MASTER_PID" || rc=$? + MASTER_PID="" + return "$rc" +} + +cleanup() { + local rc=$? + trap - EXIT + if [[ -n "$TEST_PID" ]]; then + # GNU timeout owns a process group containing the test. + kill -KILL -- "-$TEST_PID" 2>/dev/null || true + wait "$TEST_PID" 2>/dev/null || true + fi + if [[ -n "$MASTER_PID" ]]; then + if ! stop_master 5; then + if [[ -n "$MASTER_PID" ]]; then + kill -KILL "$MASTER_PID" 2>/dev/null || true + wait "$MASTER_PID" 2>/dev/null || true + fi + fi + fi + nof_target_stop + if (( rc != 0 )); then + echo "FAILED: NoF heartbeat recovery (logs: $LOG_DIR)" >&2 + tail -n 60 "$LOG_DIR/client.log" "$LOG_DIR/master.log" "$LOG_DIR/target-1.log" \ + "$LOG_DIR/target-2.log" 2>/dev/null >&2 || true + fi + exit "$rc" +} + +[[ "$TIMEOUT_SEC" =~ ^[1-9][0-9]*$ ]] && (( TIMEOUT_SEC > 30 )) || { + echo "TIMEOUT_SEC must be greater than 30" >&2; exit 2 +} +[[ -x "$MASTER_BIN" && -x "$TEST_BIN" && -x "$SPDK_DIR/build/bin/nvmf_tgt" ]] || { + echo "Build mooncake_master, nof_heartbeat_recovery_e2e (USE_NOF=ON) and SPDK nvmf_tgt first" >&2 + exit 2 +} +mkdir -p "$LOG_DIR" +[[ -z "$(ls -A "$LOG_DIR")" ]] || { echo "LOG_DIR must be empty" >&2; exit 2; } +LOG_DIR=$(cd -- "$LOG_DIR" && pwd) +RPC_SOCKET="$LOG_DIR/spdk.sock" +[[ ${#RPC_SOCKET} -lt 104 ]] || { + echo "LOG_DIR path is too long for an RPC socket" >&2; exit 2 +} +trap cleanup EXIT +trap 'exit 130' INT +trap 'exit 143' TERM +export LD_LIBRARY_PATH="$BUILD_DIR/mooncake-common${LD_LIBRARY_PATH:+:$LD_LIBRARY_PATH}" +export MC_RPC_PROTOCOL=tcp +export MC_RPC_TIMEOUT_MS=2000 +export MC_RPC_CONNECT_TIMEOUT_MS=2000 + +start_target() { + nof_target_start "$LOG_DIR/target-$1.log" || return + nof_target_add_malloc_ns 1 32 4096 || return + nof_target_add_malloc_ns 2 32 4096 || return + nof_target_start_listener +} + +wait_for_pattern() { + local pattern=$1 deadline=$((SECONDS + TIMEOUT_SEC)) + until grep -q "$pattern" "$LOG_DIR/client.log"; do + kill -0 "$TEST_PID" 2>/dev/null && + kill -0 "$MASTER_PID" 2>/dev/null && (( SECONDS < deadline )) || return 1 + sleep 0.1 + done +} + +start_target 1 +ENDPOINT="trtype:TCP adrfam:IPv4 traddr:$TARGET_HOST trsvcid:$TARGET_PORT subnqn:$TARGET_NQN" +MC_NVME_FABRICS_CONNECT_TIMEOUT_US=500000 \ + "$MASTER_BIN" --rpc_address="$MASTER_HOST" --rpc_port="$MASTER_PORT" \ + --metrics_port="$METRICS_PORT" --enable_ha=false \ + --enable_http_metadata_server=false --logtostderr=true --v=1 \ + --nof_heartbeat_interval_sec=10 --nof_heartbeat_probe_timeout_ms=1000 \ + --nof_heartbeat_failures_threshold=3 >"$LOG_DIR/master.log" 2>&1 & +MASTER_PID=$! + +# The helper only registers/queries through RPC and observes Master logs. +# It never creates MasterService or drives SPDK recovery. +timeout --kill-after=5 "$((TIMEOUT_SEC * 6))" "$TEST_BIN" \ + --master_server="$MASTER_RPC" --endpoint="$ENDPOINT" \ + --control_dir="$LOG_DIR" --timeout_sec="$TIMEOUT_SEC" \ + >"$LOG_DIR/client.log" 2>&1 & +TEST_PID=$! +wait_for_pattern '^PASS initial_heartbeats$' + +kill -STOP "$TARGET_PID" +deadline=$((SECONDS + TIMEOUT_SEC)) +until grep -q '^State:.*T' "/proc/$TARGET_PID/status"; do + (( SECONDS < deadline )) || exit 1 + sleep 0.01 +done +touch "$LOG_DIR/target-paused" +wait_for_pattern '^PASS probe_timeout$' +nof_target_stop + +# Recreate the target with the same endpoint and namespace configuration. +start_target 2 +touch "$LOG_DIR/target-restarted" +wait_for_pattern '^PASS heartbeats_recovered$' + +nof_target_stop +touch "$LOG_DIR/target-down" +wait_for_pattern '^PASS heartbeat_checks$' +wait "$TEST_PID" +TEST_PID="" +stop_master "$TIMEOUT_SEC" +echo "PASS master_shutdown" +echo "PASSED: Master NoF heartbeat recovery (logs: $LOG_DIR)"