diff --git a/sdk/c/libdingofs.h b/sdk/c/libdingofs.h index 62f3756b7..5e9886115 100644 --- a/sdk/c/libdingofs.h +++ b/sdk/c/libdingofs.h @@ -168,6 +168,9 @@ uintptr_t dingofs_new(void); * dingofs_delete() — free all memory associated with `h`. * * Behaviour is undefined if the filesystem is still mounted. + * Stop submitting API calls and join all callers before unmount/delete. + * Admission drain is not a lifetime barrier for the opaque handle or for + * calls rejected during shutdown. * Passing 0 is a no-op. */ void dingofs_delete(uintptr_t h); @@ -299,6 +302,8 @@ int dingofs_mount_nolog(uintptr_t h, * dingofs_umount() — flush pending writes and disconnect from the MDS. * * All open file descriptors are implicitly flushed and closed. + * The caller must first stop submitting calls and join all threads using `h`. + * Do not call from an operation callback; it would wait for that operation. * Returns 0 on success, -errno on failure. */ int dingofs_umount(uintptr_t h); diff --git a/sdk/python/dingofs/client.py b/sdk/python/dingofs/client.py index 1280557b9..3689eed65 100644 --- a/sdk/python/dingofs/client.py +++ b/sdk/python/dingofs/client.py @@ -270,6 +270,10 @@ def start(self, config: Config) -> None: def stop(self) -> None: """Unmount the filesystem and release all resources. + Stop submitting operations and join all threads/callbacks using this + client before calling stop() or releasing it. Admission drain does not + protect callers still entering the wrapper during shutdown. + Raises: DingofsError: if unmounting fails. """ diff --git a/sdk/shim/binding_client.h b/sdk/shim/binding_client.h index 407ed9bea..a2fe5f9d0 100644 --- a/sdk/shim/binding_client.h +++ b/sdk/shim/binding_client.h @@ -72,6 +72,10 @@ struct OptionInfo { // BindingClient wraps ClientSession for use from language bindings. // It handles gflags/logging setup before delegating to ClientSession. +// The caller owns concurrency and lifetime: stop submitting calls and join all +// operation/callback threads before Stop or destruction. Session admission +// drain does not join rejected callers or protect this wrapper's lifetime; +// Stop may also shut down the logging runtime owned by this instance. class BindingClient { public: BindingClient(); diff --git a/src/client/vfs/CMakeLists.txt b/src/client/vfs/CMakeLists.txt index a4fbb0dac..6210b291a 100644 --- a/src/client/vfs/CMakeLists.txt +++ b/src/client/vfs/CMakeLists.txt @@ -24,6 +24,7 @@ add_subdirectory(compaction) add_library(vfs_lib vfs_impl.cc client_session.cc + operation_tracker.cc access_log.cc ) diff --git a/src/client/vfs/client_session.cc b/src/client/vfs/client_session.cc index bc12b8a78..71b67f814 100644 --- a/src/client/vfs/client_session.cc +++ b/src/client/vfs/client_session.cc @@ -45,6 +45,7 @@ #include "common/options/cache.h" #include "common/options/client.h" #include "common/status.h" +#include "common/sync_point.h" #include "common/trace/trace_manager.h" #include "common/types.h" #include "fmt/format.h" @@ -308,35 +309,19 @@ ClientSession::ClientSession() = default; ClientSession::~ClientSession() { Stop(/*handover=*/false); } ClientSession::OperationLease::~OperationLease() { - if (owner_ != nullptr) owner_->ReleaseOperation(); + if (lease_) g_active_public_operations << -1; } std::optional ClientSession::TryAcquireOperation() { - { - std::lock_guard lock(lifecycle_mutex_); - if (lifecycle_state_ != LifecycleState::kRunning) { - g_rejected_public_operations << 1; - return std::nullopt; - } - - ++active_public_operations_; + auto lease = operations_.TryEnter(); + if (!lease) { + g_rejected_public_operations << 1; + return std::nullopt; } g_active_public_operations << 1; - return OperationLease(this); -} - -void ClientSession::ReleaseOperation() { - g_active_public_operations << -1; - - std::lock_guard lock(lifecycle_mutex_); - CHECK_GT(active_public_operations_, 0); - --active_public_operations_; - if (active_public_operations_ == 0 && - lifecycle_state_ == LifecycleState::kQuiescing) { - lifecycle_cv_.notify_all(); - } + return OperationLease(std::move(*lease)); } Status ClientSession::FinishStartFailure(const Status& status) { @@ -371,6 +356,7 @@ Status ClientSession::Start(const DingofsConfig& config, int upgrade_from_pid) { } lifecycle_state_ = LifecycleState::kStarting; } + TEST_SYNC_POINT_CALLBACK("ClientSession::Starting", this); if (config.fs_name.empty()) { return FinishStartFailure(Status::InvalidParam("fs_name is empty")); @@ -467,10 +453,12 @@ Status ClientSession::Start(const DingofsConfig& config, int upgrade_from_pid) { uid_ = dingofs::Helper::GetOriginalUid(); gid_ = dingofs::Helper::GetOriginalGid(); + TEST_SYNC_POINT_CALLBACK("ClientSession::BeforePublishRunning", this); { std::lock_guard lock(lifecycle_mutex_); lifecycle_state_ = LifecycleState::kRunning; stop_status_ = Status::OK(); + operations_.OpenOnce(); } lifecycle_cv_.notify_all(); return Status::OK(); @@ -499,6 +487,7 @@ Status ClientSession::Stop(bool handover) { if (lifecycle_state_ == LifecycleState::kQuiescing) { const bool same_stop_mode = stop_handover_ == handover; lifecycle_cv_.wait(lock, [this]() { + TEST_SYNC_POINT_CALLBACK("ClientSession::StopWaiting", this); return lifecycle_state_ == LifecycleState::kStopped; }); if (!same_stop_mode) { @@ -509,17 +498,11 @@ Status ClientSession::Stop(bool handover) { } CHECK(lifecycle_state_ == LifecycleState::kRunning); + operations_.Close(); lifecycle_state_ = LifecycleState::kQuiescing; stop_handover_ = handover; - while (active_public_operations_ != 0) { - if (lifecycle_cv_.wait_for(lock, std::chrono::seconds(30)) == - std::cv_status::timeout) { - LOG(ERROR) << fmt::format( - "VFS Stop still waiting for {} public operation(s) to drain", - active_public_operations_); - } - } } + operations_.WaitForDrain(); LOG(INFO) << fmt::format("stopping vfs, handover({}).", handover); diff --git a/src/client/vfs/client_session.h b/src/client/vfs/client_session.h index c4eefd57b..0a6ea1743 100644 --- a/src/client/vfs/client_session.h +++ b/src/client/vfs/client_session.h @@ -26,6 +26,7 @@ #include #include +#include "client/vfs/operation_tracker.h" #include "client/vfs/vfs.h" #include "common/meta.h" #include "common/status.h" @@ -70,8 +71,9 @@ class ClientSession { public: ClientSession(); - // Blocks until admitted operations and Core teardown complete. The owner - // must not destroy this object concurrently with another method call. + // Stop drains admitted runtime access, not late rejection/notification tails. + // The owner must close external entrypoints and join every caller before + // destruction, including callers rejected during or after Stop. ~ClientSession(); // Normal start: upgrade_from_pid = 0. @@ -83,6 +85,8 @@ class ClientSession { // Stop the VFS. With handover=true, skip MDS unmount and persist state for // the new process. It is idempotent after success, so post-exit teardown does // not stop/dump twice. + // Never call Stop from an operation/callback holding this session's lease: + // it would wait for itself. Stop does not replace the owner's caller join. Status Stop(bool handover = false); Status GetInfo(std::string* info); @@ -194,22 +198,20 @@ class ClientSession { OperationLease(const OperationLease&) = delete; OperationLease& operator=(const OperationLease&) = delete; - OperationLease(OperationLease&& other) noexcept - : owner_(std::exchange(other.owner_, nullptr)) {} + OperationLease(OperationLease&& other) noexcept = default; ~OperationLease(); private: friend class ClientSession; - explicit OperationLease(ClientSession* owner) : owner_(owner) {} + explicit OperationLease(OperationTracker::Lease lease) + : lease_(std::move(lease)) {} - ClientSession* owner_; + OperationTracker::Lease lease_; }; std::optional TryAcquireOperation(); - void ReleaseOperation(); - Status FinishStartFailure(const Status& status); bool Dump(); @@ -219,7 +221,7 @@ class ClientSession { mutable std::mutex lifecycle_mutex_; std::condition_variable lifecycle_cv_; LifecycleState lifecycle_state_{LifecycleState::kCreated}; - uint64_t active_public_operations_{0}; + OperationTracker operations_; Status stop_status_; bool stop_handover_{false}; bool trace_started_{false}; diff --git a/src/client/vfs/operation_tracker.cc b/src/client/vfs/operation_tracker.cc new file mode 100644 index 000000000..025df3ade --- /dev/null +++ b/src/client/vfs/operation_tracker.cc @@ -0,0 +1,72 @@ +/* + * Copyright (c) 2026 dingodb.com, Inc. All Rights Reserved + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include "client/vfs/operation_tracker.h" + +#include + +namespace dingofs { +namespace client { +namespace { + +std::atomic next_tracker_thread_index{0}; + +} // namespace + +uint64_t OperationTracker::AllocateThreadIndex() { + const auto index = + next_tracker_thread_index.fetch_add(1, std::memory_order_seq_cst); + CHECK_NE(index, std::numeric_limits::max()) + << "Session operation thread index overflow"; + return index; +} + +void OperationTracker::OpenOnce() { + CHECK(!opened_) << "Operation tracker cannot reopen"; + opened_ = true; + epoch_.store(2, std::memory_order_seq_cst); +} + +bool OperationTracker::IsDrained() const { + for (const auto& counter : counts_) { + if (counter.value.load(std::memory_order_seq_cst) != 0) return false; + } + return true; +} + +void OperationTracker::WaitForDrain() { + CHECK_NE(epoch_.load(std::memory_order_seq_cst) & 1, 0) + << "Close the operation tracker before draining"; + std::unique_lock lock(drain_mutex_); + // Accepted operations can only disappear after Close. Late tentative + // increments need not be frozen: their second epoch check must reject them. + while (!IsDrained()) { + TEST_SYNC_POINT_CALLBACK("OperationTracker::BeforeWait", this); + if (drain_cv_.wait_for(lock, std::chrono::seconds(30)) == + std::cv_status::timeout && + !IsDrained()) { + unsigned busy_slots = 0; + for (const auto& counter : counts_) { + if (counter.value.load(std::memory_order_seq_cst) != 0) ++busy_slots; + } + LOG(ERROR) << "VFS Stop still waiting for public operations to drain in " + << busy_slots << " operation tracker slot(s)"; + } + } +} + +} // namespace client +} // namespace dingofs diff --git a/src/client/vfs/operation_tracker.h b/src/client/vfs/operation_tracker.h new file mode 100644 index 000000000..1a73986a8 --- /dev/null +++ b/src/client/vfs/operation_tracker.h @@ -0,0 +1,191 @@ +/* + * Copyright (c) 2026 dingodb.com, Inc. All Rights Reserved + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#ifndef DINGOFS_CLIENT_VFS_OPERATION_TRACKER_H_ +#define DINGOFS_CLIENT_VFS_OPERATION_TRACKER_H_ + +#include + +#include +#include +#include +#include +#include +#include +#include +#include + +#include "common/sync_point.h" + +namespace dingofs { +namespace client { + +// One-shot admission tracker. The lifecycle control plane serializes OpenOnce +// against Close; only the elected Stop owner calls Close and WaitForDrain. +// The tracker starts closed and cannot reopen after its one successful +// OpenOnce. +// +// A lease must cover ALL protected runtime accesses, including their tails. +// No operation or callback holding a lease may synchronously reenter Stop or +// WaitForDrain: that would wait for its own lease. The tracker invokes no +// production callbacks and Leave never acquires a lifecycle lock. +// +// WaitForDrain protects admitted runtime accesses, NOT this object's lifetime: +// a rejected TryEnter can still roll back a late tentative increment, and a +// Leave can still be notifying after the scan sees zero. The external owner +// must close entry points and join ALL callers and notification tails before +// destroying the tracker. A zero-counter destructor check cannot replace +// joining. +class OperationTracker { + public: + static constexpr unsigned kSlotCount = 64; + + class Lease { + public: + Lease(const Lease&) = delete; + Lease& operator=(const Lease&) = delete; + + Lease(Lease&& other) noexcept + : tracker_(std::exchange(other.tracker_, nullptr)), + slot_(other.slot_) {} + + Lease& operator=(Lease&& other) noexcept { + if (this != &other) { + Release(); + tracker_ = std::exchange(other.tracker_, nullptr); + slot_ = other.slot_; + } + return *this; + } + + ~Lease() { Release(); } + + explicit operator bool() const noexcept { return tracker_ != nullptr; } + + private: + friend class OperationTracker; + + Lease(OperationTracker* tracker, unsigned slot) noexcept + : tracker_(tracker), slot_(slot) {} + + void Release() noexcept { + if (tracker_ != nullptr) { + TEST_SYNC_POINT_CALLBACK("OperationTracker::BeforeLeaseRelease", + tracker_); + tracker_->Leave(slot_); + tracker_ = nullptr; + } + } + + OperationTracker* tracker_; + unsigned slot_; + }; + + OperationTracker() = default; + OperationTracker(const OperationTracker&) = delete; + OperationTracker& operator=(const OperationTracker&) = delete; + + void OpenOnce(); + + std::optional TryEnter() { + const auto before = epoch_.load(std::memory_order_seq_cst); + if ((before & 1) != 0) return std::nullopt; + + const auto slot = ThreadSlot(); + TEST_SYNC_POINT_CALLBACK("OperationTracker::AfterFirstRead", this); + const auto old = + counts_[slot].value.fetch_add(1, std::memory_order_seq_cst); + CHECK_NE(old, std::numeric_limits::max()) + << "Operation counter overflow"; + TEST_SYNC_POINT_CALLBACK("OperationTracker::AfterIncrement", this); + const auto after = epoch_.load(std::memory_order_seq_cst); + if (after != before || (after & 1) != 0) { + // A failed second check owns the same decrement/notification duty as an + // admitted lease, even if the owner already observed a drained tracker. + Leave(slot); + return std::nullopt; + } + TEST_SYNC_POINT_CALLBACK("OperationTracker::AfterAdmission", this); + return Lease(this, slot); + } + + void Close() { epoch_.fetch_or(1, std::memory_order_seq_cst); } + + // Call after Close, without the lifecycle mutex. Timeouts only diagnose; + // they never authorize destruction of runtime still protected by a lease. + void WaitForDrain(); + +#ifndef NDEBUG + // Only this OS thread is affected; production always uses its cached slot. + static void SetTestSlot(unsigned slot) { + CHECK_LT(slot, kSlotCount); + test_slot_ = slot; + } + static void ClearTestSlot() { test_slot_ = kSlotCount; } +#endif + + private: + friend class OperationTrackerTestPeer; + + static uint64_t AllocateThreadIndex(); + + static unsigned ThreadSlot() { +#ifndef NDEBUG + if (test_slot_ != kSlotCount) return test_slot_; +#endif + // Shared across tracker instances, allocated once per OS thread, not once + // per operation. A slot is not a CPU, worker, or bthread identity. + static thread_local const unsigned slot = + static_cast(AllocateThreadIndex() & (kSlotCount - 1)); + return slot; + } + + void Leave(unsigned slot) { + const auto old = + counts_[slot].value.fetch_sub(1, std::memory_order_seq_cst); + CHECK_GT(old, 0) << "Operation counter underflow or double release"; + TEST_SYNC_POINT_CALLBACK("OperationTracker::AfterDecrement", this); + // Read the epoch AFTER decrementing. Reusing a prior open observation can + // miss the final notification when Close races with this release. + if (old == 1 && (epoch_.load(std::memory_order_seq_cst) & 1) != 0) { + TEST_SYNC_POINT_CALLBACK("OperationTracker::BeforeNotify", this); + std::lock_guard lock(drain_mutex_); + drain_cv_.notify_all(); + } + } + + bool IsDrained() const; + + struct alignas(64) Counter { + std::atomic value{0}; + }; + static_assert(sizeof(Counter) == 64, + "Each tracker counter needs one cacheline"); + + alignas(64) std::atomic epoch_{1}; + std::array counts_{}; + std::mutex drain_mutex_; + std::condition_variable drain_cv_; + bool opened_{false}; // Accessed only by the serialized lifecycle owner. +#ifndef NDEBUG + inline static thread_local unsigned test_slot_ = kSlotCount; +#endif +}; + +} // namespace client +} // namespace dingofs + +#endif // DINGOFS_CLIENT_VFS_OPERATION_TRACKER_H_ diff --git a/src/client/vfs/vfs_impl.h b/src/client/vfs/vfs_impl.h index 7f5b46dc2..054165332 100644 --- a/src/client/vfs/vfs_impl.h +++ b/src/client/vfs/vfs_impl.h @@ -35,6 +35,7 @@ namespace dingofs { namespace client { +class ClientSessionLifecycleTest; namespace vfs { class VFSImpl : public VFS { @@ -144,6 +145,7 @@ class VFSImpl : public VFS { private: friend class VFSImplTest; + friend class ::dingofs::client::ClientSessionLifecycleTest; // Test-only constructor: inject a pre-built VFSHub. VFSImpl(std::unique_ptr hub, TraceManager& trace_manager); diff --git a/test/unit/client/vfs/CMakeLists.txt b/test/unit/client/vfs/CMakeLists.txt index e553a771f..e9852e076 100644 --- a/test/unit/client/vfs/CMakeLists.txt +++ b/test/unit/client/vfs/CMakeLists.txt @@ -23,6 +23,7 @@ add_subdirectory(metasystem) add_library(test_vfs_impl test_vfs_impl.cc test_client_session_lifecycle.cc + test_operation_tracker.cc ) target_link_libraries(test_vfs_impl vfs_lib diff --git a/test/unit/client/vfs/test_base.h b/test/unit/client/vfs/test_base.h index a095d2b7d..75a299b22 100644 --- a/test/unit/client/vfs/test_base.h +++ b/test/unit/client/vfs/test_base.h @@ -240,7 +240,7 @@ class VFSTestBase : public ::testing::Test { FLAGS_vfs_meta_access_logging = false; // Default context for tests - ctx_ = std::make_shared("test"); + ctx_ = std::make_shared<::dingofs::Context>("test"); } ~VFSTestBase() override { diff --git a/test/unit/client/vfs/test_client_session_lifecycle.cc b/test/unit/client/vfs/test_client_session_lifecycle.cc index 756cdc8cf..32ca8db8b 100644 --- a/test/unit/client/vfs/test_client_session_lifecycle.cc +++ b/test/unit/client/vfs/test_client_session_lifecycle.cc @@ -5,21 +5,40 @@ * you may not use this file except in compliance with the License. */ +#include +#include +#include +#include #include #include +#include +#include +#include #include #include +#include +#include +#include +#include #include #include +#include +#include +#include #include +#include "client/vfs/access_log.h" #include "client/vfs/client_session.h" +#include "client/vfs/vfs_impl.h" #include "common/metrics/client/client.h" #include "common/options/client.h" +#include "common/sync_point.h" #include "common/trace/trace_manager.h" +#include "test/unit/client/vfs/test_base.h" #include "utils/scoped_cleanup.h" +DECLARE_string(log_dir); namespace dingofs { namespace client { @@ -28,6 +47,123 @@ using ::testing::Invoke; using ::testing::Return; namespace { +enum class TestPoint { + kAfterFirstRead, + kAfterIncrement, + kAfterAdmission, + kBeforeWait, + kAfterDecrement, + kBeforeNotify, + kStarting, + kBeforePublishRunning, + kBeforeLeaseRelease, + kStopWaiting, +}; +constexpr auto kWaitLimit = std::chrono::seconds(5); +constexpr auto kBlockedCheck = std::chrono::milliseconds(50); + +class Event { + public: + void Signal() { + std::lock_guard lock(mutex_); + ready_ = true; + cv_.notify_all(); + } + + bool Wait() { + std::unique_lock lock(mutex_); + return cv_.wait_for(lock, kWaitLimit, [this] { return ready_; }); + } + + private: + std::mutex mutex_; + std::condition_variable cv_; + bool ready_{false}; +}; + +// Every staged call has a cleanup guard that releases its pause and joins it, +// including on an ASSERT failure. A timeout fails the test, never forces Stop. +class Pause { + public: + void Block() { + if (visited_.exchange(true)) return; + entered.Signal(); + std::unique_lock lock(mutex_); + cv_.wait(lock, [this] { return released_; }); + } + + void Release() { + std::lock_guard lock(mutex_); + released_ = true; + cv_.notify_all(); + } + + Event entered; + + private: + std::atomic visited_{false}; + std::mutex mutex_; + std::condition_variable cv_; + bool released_{false}; +}; + +struct OperationHooks { + explicit OperationHooks(TestPoint target) : target(target) {} + + static void Call(void* arg, TestPoint point) { + auto& hooks = *static_cast(arg); + if (point == TestPoint::kBeforeWait) hooks.waiting.Signal(); + if (point == hooks.target) hooks.pause.Block(); + } + + TestPoint target; + Pause pause; + Event waiting; +}; + +int64_t Metric(const char* name) { + const std::string value = bvar::Variable::describe_exposed(name); + EXPECT_FALSE(value.empty()) << name; + return value.empty() ? 0 : std::stoll(value); +} + +int64_t ActiveOperations() { return Metric("vfs_active_public_operations"); } + +int64_t RejectedOperations() { + return Metric("vfs_rejected_public_operations_total"); +} + +class PausingLogSink final : public spdlog::sinks::base_sink { + public: + Pause pause; + + private: + void sink_it_(const spdlog::details::log_msg&) override { pause.Block(); } + void flush_() override {} +}; + +// Reuse the component fixture's real handles, readers, and executors. The +// mocked hub and metadata only expose the existing dependency boundaries. +class RealVFSResources : public vfs::test::VFSTestBase { + public: + void TestBody() override {} + + std::unique_ptr TakeHub() { return std::move(hub_uptr_); } + vfs::test::MockVFSHub& Hub() { return *mock_hub_; } + vfs::test::MockMetaSystem& Meta() { return *mock_meta_system_; } + vfs::HandleManager& Handles() { return *handle_manager_; } + vfs::ReaderRegistry* Readers() { return reader_registry_.get(); } + + void StopResources() { + EXPECT_TRUE(handle_manager_->Stop().ok()); + writer_table_->Stop(); + write_background_executor_->Stop(); + flush_executor_->Stop(); + read_executor_->Stop(); + read_cleanup_executor_->Stop(); + cb_executor_->Stop(); + } +}; class MockLifecycleVFS : public vfs::VFS { public: @@ -114,10 +250,6 @@ class MockLifecycleVFS : public vfs::VFS { class ClientSessionLifecycleTest : public ::testing::Test { protected: void SetUp() override { - previous_access_logging_ = FLAGS_vfs_access_logging; - previous_attr_timeout_ = FLAGS_fuse_attr_cache_timeout_s; - previous_entry_timeout_ = FLAGS_fuse_entry_cache_timeout_s; - previous_max_name_length_ = FLAGS_vfs_meta_max_name_length; FLAGS_vfs_access_logging = false; FLAGS_fuse_attr_cache_timeout_s = 1; FLAGS_fuse_entry_cache_timeout_s = 2; @@ -128,13 +260,30 @@ class ClientSessionLifecycleTest : public ::testing::Test { core_ = core.get(); session_->vfs_ = std::move(core); session_->lifecycle_state_ = ClientSession::LifecycleState::kRunning; + session_->operations_.OpenOnce(); } void TearDown() override { - FLAGS_vfs_access_logging = previous_access_logging_; - FLAGS_fuse_attr_cache_timeout_s = previous_attr_timeout_; - FLAGS_fuse_entry_cache_timeout_s = previous_entry_timeout_; - FLAGS_vfs_meta_max_name_length = previous_max_name_length_; +#ifndef NDEBUG + SyncPoint::GetInstance()->DisableProcessing(); + SyncPoint::GetInstance()->ClearAllCallBacks(); +#endif + if (session_ != nullptr) { + // Test-body captures are already gone. Verify unmet expectations before + // teardown, then discard actions so a failed staging assertion cannot + // invoke callbacks borrowing those expired locals. + if (core_ != nullptr) { + ::testing::Mock::VerifyAndClearExpectations(core_); + } + if (resources_ != nullptr) { + ::testing::Mock::VerifyAndClearExpectations(&resources_->Hub()); + ::testing::Mock::VerifyAndClearExpectations(&resources_->Meta()); + } + session_->Stop(); + if (resources_ != nullptr) resources_->StopResources(); + session_.reset(); + } + resources_.reset(); } void SetTraceStarted(bool started) { session_->trace_started_ = started; } @@ -145,23 +294,154 @@ class ClientSessionLifecycleTest : public ::testing::Test { session_->client_metrics_ = std::make_unique(); } + +#ifndef NDEBUG + void InstallHooks(OperationHooks& hooks) { + static constexpr std::pair points[] = { + {"OperationTracker::AfterFirstRead", TestPoint::kAfterFirstRead}, + {"OperationTracker::AfterIncrement", TestPoint::kAfterIncrement}, + {"OperationTracker::AfterAdmission", TestPoint::kAfterAdmission}, + {"OperationTracker::BeforeWait", TestPoint::kBeforeWait}, + {"OperationTracker::AfterDecrement", TestPoint::kAfterDecrement}, + {"OperationTracker::BeforeNotify", TestPoint::kBeforeNotify}, + {"OperationTracker::BeforeLeaseRelease", + TestPoint::kBeforeLeaseRelease}, + {"ClientSession::Starting", TestPoint::kStarting}, + {"ClientSession::BeforePublishRunning", + TestPoint::kBeforePublishRunning}, + {"ClientSession::StopWaiting", TestPoint::kStopWaiting}, + }; + auto* sync = SyncPoint::GetInstance(); + for (const auto& [name, point] : points) { + void* expected = point == TestPoint::kStarting || + point == TestPoint::kBeforePublishRunning || + point == TestPoint::kStopWaiting + ? static_cast(session_.get()) + : static_cast(&session_->operations_); + sync->SetCallBack(name, [&hooks, point, expected](void* arg) { + if (arg == expected) OperationHooks::Call(&hooks, point); + }); + } + sync->EnableProcessing(); + } + + void RunLocalStart(bool concurrent_stop) { + char temp[] = "/tmp/dingofs-session-start-XXXXXX"; + ASSERT_NE(mkdtemp(temp), nullptr); + const char* old_base = std::getenv("DINGOFS_BASE_DIR"); + const std::optional previous_base = + old_base == nullptr ? std::nullopt + : std::make_optional(old_base); + auto restore = MakeScopedCleanup([&] { + session_.reset(); + if (previous_base) { + setenv("DINGOFS_BASE_DIR", previous_base->c_str(), 1); + } else { + unsetenv("DINGOFS_BASE_DIR"); + } + std::filesystem::remove_all(temp); + }); + ASSERT_EQ(setenv("DINGOFS_BASE_DIR", temp, 1), 0); + FLAGS_log_dir = temp; + FLAGS_enable_trace = false; + FLAGS_vfs_bthread_worker_num = 0; + FLAGS_vfs_dummy_server_port = 0; + FLAGS_vfs_write_buffer_total_mb = 64; + FLAGS_vfs_write_buffer_page_size = 4096; + FLAGS_vfs_read_buffer_total_mb = 64; + FLAGS_vfs_compact_buffer_total_mb = 64; + // Existing IO-isolation mode avoids starting an unrelated cache stack. + // Metadata, VFSHub, VFSImpl, and ClientSession Start/Stop remain real. + FLAGS_vfs_use_fake_block_store = true; + FLAGS_vfs_meta_access_logging = false; + DingofsConfig config; + config.fs_name = "session-start"; + config.mount_point = std::string(temp) + "/mount"; + config.metasystem_type = "local"; + config.storage_info = std::string("storage=file&path=") + temp + "/blocks"; + + if (!concurrent_stop) { + const auto status = session_->Start(config); + ASSERT_TRUE(status.ok()) << status.ToString(); + Attr attr; + EXPECT_TRUE(session_->GetAttr(Context{0, 0, 0, 0}, kRootIno, &attr).ok()); + EXPECT_EQ(attr.ino, kRootIno); + EXPECT_TRUE(session_->Stop().ok()); + EXPECT_TRUE( + session_->GetAttr(Context{0, 0, 0, 0}, kRootIno, &attr).IsStop()); + return; + } + + OperationHooks hooks(TestPoint::kBeforePublishRunning); + InstallHooks(hooks); + auto start = + std::async(std::launch::async, [&] { return session_->Start(config); }); + auto join = MakeScopedCleanup([&] { + hooks.pause.Release(); + if (start.valid()) start.wait(); + SyncPoint::GetInstance()->DisableProcessing(); + SyncPoint::GetInstance()->ClearAllCallBacks(); + }); + ASSERT_TRUE(hooks.pause.entered.Wait()); + std::string info; + EXPECT_TRUE(session_->GetInfo(&info).IsStop()); + Event stop_called; + auto stop = std::async(std::launch::async, [&] { + stop_called.Signal(); + return session_->Stop(); + }); + auto release_stop = MakeScopedCleanup([&] { hooks.pause.Release(); }); + ASSERT_TRUE(stop_called.Wait()); + EXPECT_EQ(stop.wait_for(kBlockedCheck), std::future_status::timeout); + hooks.pause.Release(); + const auto start_result = start.get(); + EXPECT_TRUE(start_result.ok()) << start_result.ToString(); + const auto stop_result = stop.get(); + EXPECT_TRUE(stop_result.ok()) << stop_result.ToString(); + EXPECT_TRUE(session_->GetInfo(&info).IsStop()); + EXPECT_TRUE(session_->Stop().ok()); + } +#endif + + auto AcquireOperation() { return session_->TryAcquireOperation(); } + + void ResetCreatedSession(bool install_core = false) { + EXPECT_CALL(*core_, Stop(false)).WillOnce(Return(Status::OK())); + session_.reset(); + session_ = std::make_unique(); + core_ = nullptr; + if (install_core) { + session_->trace_manager_ = std::make_unique(); + auto core = std::make_unique(); + core_ = core.get(); + session_->vfs_ = std::move(core); + } + } + + RealVFSResources& UseRealVFS() { + InitializeMetrics(); + resources_ = std::make_unique(); + ON_CALL(resources_->Hub(), GetTraceManager()) + .WillByDefault(Return(session_->trace_manager_.get())); + EXPECT_CALL(resources_->Hub(), GetTraceManager()) + .Times(::testing::AnyNumber()); + session_->vfs_.reset( + new vfs::VFSImpl(resources_->TakeHub(), *session_->trace_manager_)); + core_ = nullptr; + return *resources_; + } + + gflags::FlagSaver flag_saver_; + std::unique_ptr resources_; std::unique_ptr session_; MockLifecycleVFS* core_{nullptr}; - bool previous_access_logging_{true}; - uint32_t previous_attr_timeout_{0}; - uint32_t previous_entry_timeout_{0}; - uint32_t previous_max_name_length_{0}; }; TEST_F(ClientSessionLifecycleTest, StopWaitsForAdmittedOperation) { - std::promise entered; - std::promise release; - auto release_future = release.get_future(); + Pause admitted; std::atomic core_stopped{false}; - EXPECT_CALL(*core_, GetInfo(_)).WillOnce(Invoke([&](std::string*) { - entered.set_value(); - release_future.wait(); + admitted.Block(); return Status::OK(); })); EXPECT_CALL(*core_, Stop(false)).WillOnce(Invoke([&](bool) { @@ -169,46 +449,48 @@ TEST_F(ClientSessionLifecycleTest, StopWaitsForAdmittedOperation) { return Status::OK(); })); - std::thread operation([&] { + auto operation = std::async(std::launch::async, [&] { std::string info; - EXPECT_TRUE(session_->GetInfo(&info).ok()); + return session_->GetInfo(&info); }); - ASSERT_EQ(entered.get_future().wait_for(std::chrono::seconds(5)), - std::future_status::ready); - + auto join = MakeScopedCleanup([&] { + admitted.Release(); + if (operation.valid()) operation.wait(); + }); + ASSERT_TRUE(admitted.entered.Wait()); auto stop_future = std::async(std::launch::async, [&] { return session_->Stop(false); }); - EXPECT_EQ(stop_future.wait_for(std::chrono::milliseconds(100)), - std::future_status::timeout); + auto release_stop = MakeScopedCleanup([&] { admitted.Release(); }); + EXPECT_EQ(stop_future.wait_for(kBlockedCheck), std::future_status::timeout); EXPECT_FALSE(core_stopped.load()); - - release.set_value(); - operation.join(); + admitted.Release(); + EXPECT_TRUE(operation.get().ok()); EXPECT_TRUE(stop_future.get().ok()); EXPECT_TRUE(core_stopped.load()); } TEST_F(ClientSessionLifecycleTest, ConcurrentStopRunsCoreStopOnce) { - std::promise stop_entered; - std::promise release_stop; - auto release_future = release_stop.get_future(); - + Pause teardown; EXPECT_CALL(*core_, Stop(false)).WillOnce(Invoke([&](bool) { - stop_entered.set_value(); - release_future.wait(); + teardown.Block(); return Status::OK(); })); - auto first = std::async(std::launch::async, [&] { return session_->Stop(false); }); - ASSERT_EQ(stop_entered.get_future().wait_for(std::chrono::seconds(5)), - std::future_status::ready); - auto second = - std::async(std::launch::async, [&] { return session_->Stop(false); }); - - EXPECT_EQ(second.wait_for(std::chrono::milliseconds(100)), - std::future_status::timeout); - release_stop.set_value(); + auto join = MakeScopedCleanup([&] { + teardown.Release(); + if (first.valid()) first.wait(); + }); + ASSERT_TRUE(teardown.entered.Wait()); + Event second_called; + auto second = std::async(std::launch::async, [&] { + second_called.Signal(); + return session_->Stop(false); + }); + auto release_second = MakeScopedCleanup([&] { teardown.Release(); }); + ASSERT_TRUE(second_called.Wait()); + EXPECT_EQ(second.wait_for(kBlockedCheck), std::future_status::timeout); + teardown.Release(); EXPECT_TRUE(first.get().ok()); EXPECT_TRUE(second.get().ok()); } @@ -284,5 +566,735 @@ TEST_F(ClientSessionLifecycleTest, StoppedSessionRejectsBeforeCoreAccess) { EXPECT_DOUBLE_EQ(session_->GetEntryTimeout(kDirectory), 2.0); } +#ifndef NDEBUG +TEST_F(ClientSessionLifecycleTest, CloseAfterFirstReadRejectsLateRegistration) { + OperationHooks hooks(TestPoint::kAfterFirstRead); + InstallHooks(hooks); + const auto active = ActiveOperations(); + const auto rejected = RejectedOperations(); + EXPECT_CALL(*core_, GetInfo(_)).Times(0); + EXPECT_CALL(*core_, Stop(false)).WillOnce(Return(Status::OK())); + + auto operation = std::async(std::launch::async, [&] { + std::string info; + return session_->GetInfo(&info); + }); + auto join = MakeScopedCleanup([&] { + hooks.pause.Release(); + if (operation.valid()) operation.wait(); + }); + ASSERT_TRUE(hooks.pause.entered.Wait()); + // No tentative count yet: Stop may finish while this caller is still alive. + EXPECT_TRUE(session_->Stop().ok()); + EXPECT_EQ(ActiveOperations(), active); + hooks.pause.Release(); + EXPECT_TRUE(operation.get().IsStop()); + EXPECT_EQ(RejectedOperations(), rejected + 1); + EXPECT_EQ(ActiveOperations(), active); +} + +TEST_F(ClientSessionLifecycleTest, + CloseAfterIncrementWaitsForRejectedRollback) { + OperationHooks hooks(TestPoint::kAfterIncrement); + InstallHooks(hooks); + const auto active = ActiveOperations(); + const auto rejected = RejectedOperations(); + std::atomic core_stopped{false}; + EXPECT_CALL(*core_, GetInfo(_)).Times(0); + EXPECT_CALL(*core_, Stop(false)).WillOnce(Invoke([&](bool) { + core_stopped = true; + return Status::OK(); + })); + + auto operation = std::async(std::launch::async, [&] { + std::string info; + return session_->GetInfo(&info); + }); + auto join = MakeScopedCleanup([&] { + hooks.pause.Release(); + if (operation.valid()) operation.wait(); + }); + ASSERT_TRUE(hooks.pause.entered.Wait()); + auto stop = std::async(std::launch::async, [&] { return session_->Stop(); }); + auto release_stop = MakeScopedCleanup([&] { hooks.pause.Release(); }); + ASSERT_TRUE(hooks.waiting.Wait()); + EXPECT_EQ(ActiveOperations(), active); + EXPECT_FALSE(core_stopped.load()); + hooks.pause.Release(); + EXPECT_TRUE(operation.get().IsStop()); + EXPECT_TRUE(stop.get().ok()); + EXPECT_EQ(RejectedOperations(), rejected + 1); + EXPECT_EQ(ActiveOperations(), active); +} + +TEST_F(ClientSessionLifecycleTest, AdmittedBeforeCoreStillHoldsStop) { + OperationHooks hooks(TestPoint::kAfterAdmission); + InstallHooks(hooks); + std::atomic core_entered{false}; + std::atomic core_stopped{false}; + EXPECT_CALL(*core_, GetInfo(_)).WillOnce(Invoke([&](std::string*) { + EXPECT_FALSE(core_stopped.load()); + core_entered = true; + return Status::OK(); + })); + EXPECT_CALL(*core_, Stop(false)).WillOnce(Invoke([&](bool) { + EXPECT_TRUE(core_entered.load()); + core_stopped = true; + return Status::OK(); + })); + auto operation = std::async(std::launch::async, [&] { + std::string info; + return session_->GetInfo(&info); + }); + auto join = MakeScopedCleanup([&] { + hooks.pause.Release(); + if (operation.valid()) operation.wait(); + }); + ASSERT_TRUE(hooks.pause.entered.Wait()); + auto stop = std::async(std::launch::async, [&] { return session_->Stop(); }); + auto release_stop = MakeScopedCleanup([&] { hooks.pause.Release(); }); + ASSERT_TRUE(hooks.waiting.Wait()); + EXPECT_FALSE(core_entered.load()); + EXPECT_FALSE(core_stopped.load()); + std::string info; + EXPECT_TRUE(session_->GetInfo(&info).IsStop()); + hooks.pause.Release(); + EXPECT_TRUE(operation.get().ok()); + EXPECT_TRUE(stop.get().ok()); +} + +TEST_F(ClientSessionLifecycleTest, QuiescingRejectsBeforeCoreAccess) { + Pause teardown; + const auto rejected = RejectedOperations(); + EXPECT_CALL(*core_, Stop(false)).WillOnce(Invoke([&](bool) { + teardown.Block(); + return Status::OK(); + })); + EXPECT_CALL(*core_, GetInfo(_)).Times(0); + auto stop = std::async(std::launch::async, [&] { return session_->Stop(); }); + auto join = MakeScopedCleanup([&] { + teardown.Release(); + if (stop.valid()) stop.wait(); + }); + ASSERT_TRUE(teardown.entered.Wait()); + std::string info; + const auto status = session_->GetInfo(&info); + EXPECT_TRUE(status.IsStop()); + EXPECT_EQ(status.ToSysErrNo(), EIO); + EXPECT_EQ(RejectedOperations(), rejected + 1); + teardown.Release(); + EXPECT_TRUE(stop.get().ok()); +} + +TEST_F(ClientSessionLifecycleTest, StartFailureCleansUpAndNeverOpensAdmission) { + ResetCreatedSession(true); + SetTraceStarted(true); + EXPECT_CALL(*core_, Start(_)).Times(0); + EXPECT_CALL(*core_, GetInfo(_)).Times(0); + EXPECT_CALL(*core_, Stop(false)) + .WillOnce(Return(Status::Internal("cleanup failed"))); + const auto rejected = RejectedOperations(); + const auto start = session_->Start(DingofsConfig{}); + EXPECT_TRUE(start.IsInvalidParam()); + EXPECT_FALSE(TraceStarted()); + std::string info; + EXPECT_TRUE(session_->GetInfo(&info).IsStop()); + EXPECT_EQ(RejectedOperations(), rejected + 1); + EXPECT_EQ(session_->Stop().ToString(), start.ToString()); + EXPECT_EQ(session_->Stop(true).ToString(), start.ToString()); + EXPECT_TRUE(session_->Start(DingofsConfig{}).IsInvalidParam()); +} + +TEST_F(ClientSessionLifecycleTest, CreatedStopPreventsAnyLaterStart) { + ResetCreatedSession(); + const auto rejected = RejectedOperations(); + std::string info; + EXPECT_TRUE(session_->GetInfo(&info).IsStop()); + EXPECT_TRUE(session_->Stop().ok()); + DingofsConfig config; + config.fs_name = "must-not-start"; + config.mount_point = "/unused"; + config.metasystem_type = "memory"; + EXPECT_TRUE(session_->Start(config).IsInvalidParam()); + EXPECT_TRUE(session_->GetInfo(&info).IsStop()); + EXPECT_TRUE(session_->Stop(true).ok()); + EXPECT_EQ(RejectedOperations(), rejected + 2); +} + +TEST_F(ClientSessionLifecycleTest, StopDuringStartingWaitsForStartFailure) { + ResetCreatedSession(true); + OperationHooks hooks(TestPoint::kStarting); + InstallHooks(hooks); + EXPECT_CALL(*core_, GetInfo(_)).Times(0); + EXPECT_CALL(*core_, Stop(false)).WillOnce(Return(Status::OK())); + auto start = std::async(std::launch::async, + [&] { return session_->Start(DingofsConfig{}); }); + auto join = MakeScopedCleanup([&] { + hooks.pause.Release(); + if (start.valid()) start.wait(); + }); + ASSERT_TRUE(hooks.pause.entered.Wait()); + Event stop_called; + auto stop = std::async(std::launch::async, [&] { + stop_called.Signal(); + return session_->Stop(); + }); + auto release_stop = MakeScopedCleanup([&] { hooks.pause.Release(); }); + ASSERT_TRUE(stop_called.Wait()); + EXPECT_EQ(stop.wait_for(kBlockedCheck), std::future_status::timeout); + std::string info; + EXPECT_TRUE(session_->GetInfo(&info).IsStop()); + hooks.pause.Release(); + const auto result = start.get(); + EXPECT_TRUE(result.IsInvalidParam()); + EXPECT_EQ(stop.get().ToString(), result.ToString()); + EXPECT_EQ(session_->Stop().ToString(), result.ToString()); +} + +TEST_F(ClientSessionLifecycleTest, RepeatedStopPreservesOriginalFailure) { + const auto failure = Status::Internal("core shutdown failed"); + EXPECT_CALL(*core_, Stop(false)).WillOnce(Return(failure)); + EXPECT_CALL(*core_, Dump(_, _)).Times(0); + EXPECT_EQ(session_->Stop().ToString(), failure.ToString()); + EXPECT_EQ(session_->Stop().ToString(), failure.ToString()); + // Once Stopped, the existing contract returns the recorded result even if + // the caller changes mode; only an overlapping conflicting Stop is invalid. + EXPECT_EQ(session_->Stop(true).ToString(), failure.ToString()); + std::string info; + EXPECT_TRUE(session_->GetInfo(&info).IsStop()); +} + +TEST_F(ClientSessionLifecycleTest, SameModeStopWaitersShareCoreFailure) { + Pause teardown; + const auto failure = Status::Internal("core shutdown failed"); + EXPECT_CALL(*core_, Stop(false)).WillOnce(Invoke([&](bool) { + teardown.Block(); + return failure; + })); + auto first = std::async(std::launch::async, [&] { return session_->Stop(); }); + auto join = MakeScopedCleanup([&] { + teardown.Release(); + if (first.valid()) first.wait(); + }); + ASSERT_TRUE(teardown.entered.Wait()); + Event second_called; + auto second = std::async(std::launch::async, [&] { + second_called.Signal(); + return session_->Stop(); + }); + auto release_second = MakeScopedCleanup([&] { teardown.Release(); }); + ASSERT_TRUE(second_called.Wait()); + EXPECT_EQ(second.wait_for(kBlockedCheck), std::future_status::timeout); + teardown.Release(); + EXPECT_EQ(first.get().ToString(), failure.ToString()); + EXPECT_EQ(second.get().ToString(), failure.ToString()); +} + +TEST_F(ClientSessionLifecycleTest, ConflictingStopWaitsForFullTeardown) { + OperationHooks hooks(TestPoint::kStopWaiting); + InstallHooks(hooks); + Pause teardown; + const auto failure = Status::Internal("first stop failed"); + EXPECT_CALL(*core_, Stop(false)).WillOnce(Invoke([&](bool) { + teardown.Block(); + return failure; + })); + EXPECT_CALL(*core_, Stop(true)).Times(0); + EXPECT_CALL(*core_, Dump(_, _)).Times(0); + auto first = std::async(std::launch::async, [&] { return session_->Stop(); }); + auto join = MakeScopedCleanup([&] { + teardown.Release(); + if (first.valid()) first.wait(); + }); + ASSERT_TRUE(teardown.entered.Wait()); + auto conflicting = + std::async(std::launch::async, [&] { return session_->Stop(true); }); + auto release_conflict = MakeScopedCleanup([&] { + hooks.pause.Release(); + teardown.Release(); + }); + // The conflicting caller has observed Quiescing while teardown is paused. + ASSERT_TRUE(hooks.pause.entered.Wait()); + hooks.pause.Release(); + teardown.Release(); + EXPECT_EQ(first.get().ToString(), failure.ToString()); + EXPECT_TRUE(conflicting.get().IsInvalidParam()); + EXPECT_EQ(session_->Stop(true).ToString(), failure.ToString()); +} + +TEST_F(ClientSessionLifecycleTest, HandoverFailureSkipsDumpAndStopsTrace) { + SetTraceStarted(true); + const auto failure = Status::Internal("handover stop failed"); + EXPECT_CALL(*core_, Stop(true)).WillOnce(Invoke([&](bool) { + EXPECT_TRUE(TraceStarted()); + return failure; + })); + EXPECT_CALL(*core_, Dump(_, _)).Times(0); + EXPECT_EQ(session_->Stop(true).ToString(), failure.ToString()); + EXPECT_FALSE(TraceStarted()); + EXPECT_EQ(session_->Stop(true).ToString(), failure.ToString()); +} + +TEST_F(ClientSessionLifecycleTest, HandoverDumpFailureNotifiesSameModeWaiter) { + Pause dump; + SetTraceStarted(true); + { + ::testing::InSequence order; + EXPECT_CALL(*core_, Stop(true)).WillOnce(Return(Status::OK())); + EXPECT_CALL(*core_, Dump(_, _)) + .WillOnce(Invoke([&](ContextSPtr, Json::Value&) { + EXPECT_TRUE(TraceStarted()); + dump.Block(); + return false; + })); + } + auto first = + std::async(std::launch::async, [&] { return session_->Stop(true); }); + auto join = MakeScopedCleanup([&] { + dump.Release(); + if (first.valid()) first.wait(); + }); + ASSERT_TRUE(dump.entered.Wait()); + Event second_called; + auto second = std::async(std::launch::async, [&] { + second_called.Signal(); + return session_->Stop(true); + }); + auto release_second = MakeScopedCleanup([&] { dump.Release(); }); + ASSERT_TRUE(second_called.Wait()); + EXPECT_EQ(second.wait_for(kBlockedCheck), std::future_status::timeout); + std::string info; + EXPECT_TRUE(session_->GetInfo(&info).IsStop()); + dump.Release(); + const auto result = first.get(); + EXPECT_TRUE(result.IsInvalidParam()); + EXPECT_EQ(second.get().ToString(), result.ToString()); + EXPECT_EQ(session_->Stop(true).ToString(), result.ToString()); + EXPECT_FALSE(TraceStarted()); +} + +TEST_F(ClientSessionLifecycleTest, MovedFromLeaseDoesNotChangeActiveOwnership) { + const auto active = ActiveOperations(); + const auto rejected = RejectedOperations(); + auto source = AcquireOperation(); + ASSERT_TRUE(source.has_value()); + EXPECT_EQ(ActiveOperations(), active + 1); + auto destination = std::move(source); + source.reset(); + EXPECT_EQ(ActiveOperations(), active + 1); + EXPECT_EQ(RejectedOperations(), rejected); + + OperationHooks hooks(TestPoint::kAfterFirstRead); + InstallHooks(hooks); + EXPECT_CALL(*core_, Stop(false)).WillOnce(Invoke([&](bool) { + EXPECT_EQ(ActiveOperations(), active); + return Status::OK(); + })); + auto stop = std::async(std::launch::async, [&] { return session_->Stop(); }); + auto release_lease = MakeScopedCleanup([&] { destination.reset(); }); + ASSERT_TRUE(hooks.waiting.Wait()); + std::string info; + EXPECT_TRUE(session_->GetInfo(&info).IsStop()); + EXPECT_EQ(RejectedOperations(), rejected + 1); + EXPECT_EQ(ActiveOperations(), active + 1); + destination.reset(); + EXPECT_TRUE(stop.get().ok()); + EXPECT_EQ(ActiveOperations(), active); +} + +TEST_F(ClientSessionLifecycleTest, + FinalTrackerReleaseStillHoldsStopAfterActiveZero) { + OperationHooks hooks(TestPoint::kBeforeLeaseRelease); + InstallHooks(hooks); + const auto active = ActiveOperations(); + std::atomic core_stopped{false}; + EXPECT_CALL(*core_, GetInfo(_)).WillOnce(Return(Status::OK())); + EXPECT_CALL(*core_, Stop(false)).WillOnce(Invoke([&](bool) { + core_stopped = true; + return Status::OK(); + })); + auto operation = std::async(std::launch::async, [&] { + std::string info; + return session_->GetInfo(&info); + }); + auto join = MakeScopedCleanup([&] { + hooks.pause.Release(); + if (operation.valid()) operation.wait(); + }); + ASSERT_TRUE(hooks.pause.entered.Wait()); + EXPECT_EQ(ActiveOperations(), active); + auto stop = std::async(std::launch::async, [&] { return session_->Stop(); }); + auto release_stop = MakeScopedCleanup([&] { hooks.pause.Release(); }); + ASSERT_TRUE(hooks.waiting.Wait()); + EXPECT_FALSE(core_stopped.load()); + hooks.pause.Release(); + EXPECT_TRUE(operation.get().ok()); + EXPECT_TRUE(stop.get().ok()); +} + +TEST_F(ClientSessionLifecycleTest, StopWaitsForUnpublishedRealHandle) { + auto& resources = UseRealVFS(); + OperationHooks hooks(TestPoint::kBeforePublishRunning); + InstallHooks(hooks); + Pause before_publish; + uint64_t internal_fh = 0; + EXPECT_CALL(resources.Meta(), Open(_, 42, O_RDONLY, _, _)) + .WillOnce(Invoke([&](ContextSPtr, Ino, int, uint64_t fh, bool*) { + internal_fh = fh; + return Status::OK(); + })); + EXPECT_CALL(resources.Hub(), GetReaderRegistry()) + .WillOnce(Invoke([&] { + before_publish.Block(); + return resources.Readers(); + })) + .RetiresOnSaturation(); + std::atomic stopped{false}; + EXPECT_CALL(resources.Hub(), Stop(false)).WillOnce(Invoke([&](bool) { + stopped = true; + resources.StopResources(); + return Status::OK(); + })); + + uint64_t fh = 0; + auto operation = std::async(std::launch::async, [&] { + bool keep_cache = false; + return session_->Open(Context{0, 0, 0, 0}, 42, O_RDONLY, &fh, &keep_cache); + }); + auto join = MakeScopedCleanup([&] { + before_publish.Release(); + if (operation.valid()) operation.wait(); + }); + ASSERT_TRUE(before_publish.entered.Wait()); + EXPECT_FALSE(resources.Handles().FindHandlerForRelease(internal_fh)); + auto stop = std::async(std::launch::async, [&] { return session_->Stop(); }); + auto release_stop = MakeScopedCleanup([&] { before_publish.Release(); }); + ASSERT_TRUE(hooks.waiting.Wait()); + EXPECT_FALSE(stopped.load()); + before_publish.Release(); + EXPECT_TRUE(operation.get().ok()); + EXPECT_TRUE(stop.get().ok()); + EXPECT_EQ(fh, internal_fh); + // Stop detaches resources, but preserves the published fh identity. + EXPECT_TRUE(resources.Handles().FindHandlerForRelease(fh)); +} + +TEST_F(ClientSessionLifecycleTest, StopWaitsAfterRealHandleErase) { + auto& resources = UseRealVFS(); + uint64_t fh = 0; + bool keep_cache = false; + ASSERT_TRUE( + session_->Open(Context{0, 0, 0, 0}, 42, O_RDONLY, &fh, &keep_cache).ok()); + ASSERT_TRUE(resources.Handles().FindHandlerForRelease(fh)); + OperationHooks hooks(TestPoint::kBeforePublishRunning); + InstallHooks(hooks); + Pause after_erase; + EXPECT_CALL(resources.Hub(), GetReaderRegistry()).WillOnce(Invoke([&] { + after_erase.Block(); + return resources.Readers(); + })); + std::atomic stopped{false}; + EXPECT_CALL(resources.Hub(), Stop(false)).WillOnce(Invoke([&](bool) { + stopped = true; + resources.StopResources(); + return Status::OK(); + })); + + auto operation = std::async(std::launch::async, [&] { + return session_->Release(Context{0, 0, 0, 0}, 42, fh); + }); + auto join = MakeScopedCleanup([&] { + after_erase.Release(); + if (operation.valid()) operation.wait(); + }); + ASSERT_TRUE(after_erase.entered.Wait()); + EXPECT_FALSE(resources.Handles().FindHandlerForRelease(fh)); + auto stop = std::async(std::launch::async, [&] { return session_->Stop(); }); + auto release_stop = MakeScopedCleanup([&] { after_erase.Release(); }); + ASSERT_TRUE(hooks.waiting.Wait()); + EXPECT_FALSE(stopped.load()); + after_erase.Release(); + EXPECT_TRUE(operation.get().ok()); + EXPECT_TRUE(stop.get().ok()); +} + +TEST_F(ClientSessionLifecycleTest, StopWaitsForRealNoHandleMetadataPath) { + auto& resources = UseRealVFS(); + OperationHooks hooks(TestPoint::kBeforePublishRunning); + InstallHooks(hooks); + Pause metadata; + std::atomic stopped{false}; + const auto expected_attr = vfs::test::MakeFileAttr(42); + EXPECT_CALL(resources.Meta(), GetAttr(_, 42, _)) + .WillOnce(Invoke([&](ContextSPtr, Ino, Attr* attr) { + metadata.Block(); + EXPECT_FALSE(stopped.load()); + *attr = expected_attr; + return Status::OK(); + })); + EXPECT_CALL(resources.Hub(), Stop(false)).WillOnce(Invoke([&](bool) { + stopped = true; + resources.StopResources(); + return Status::OK(); + })); + Attr attr; + auto operation = std::async(std::launch::async, [&] { + return session_->GetAttr(Context{0, 0, 0, 0}, 42, &attr); + }); + auto join = MakeScopedCleanup([&] { + metadata.Release(); + if (operation.valid()) operation.wait(); + }); + ASSERT_TRUE(metadata.entered.Wait()); + auto stop = std::async(std::launch::async, [&] { return session_->Stop(); }); + auto release_stop = MakeScopedCleanup([&] { metadata.Release(); }); + ASSERT_TRUE(hooks.waiting.Wait()); + EXPECT_FALSE(stopped.load()); + metadata.Release(); + EXPECT_TRUE(operation.get().ok()); + EXPECT_EQ(attr.ino, expected_attr.ino); + EXPECT_TRUE(stop.get().ok()); +} + +TEST_F(ClientSessionLifecycleTest, AccessLogDestructorRemainsInsideLease) { + InitializeMetrics(); + auto sink = std::make_shared(); + auto previous_logger = logger; + logger = std::make_shared("session-tail", sink); + auto restore_logger = + MakeScopedCleanup([&] { logger = std::move(previous_logger); }); + FLAGS_vfs_access_logging = true; + FLAGS_vfs_access_log_threshold_us = 0; + OperationHooks hooks(TestPoint::kBeforePublishRunning); + InstallHooks(hooks); + const auto active = ActiveOperations(); + std::atomic business_returned{false}; + std::atomic stopped{false}; + const auto business_status = Status::Internal("metadata failed"); + EXPECT_CALL(*core_, GetAttr(_, 42, _)) + .WillOnce(Invoke([&](ContextSPtr, Ino, Attr* attr) { + *attr = vfs::test::MakeFileAttr(42); + business_returned = true; + return business_status; + })); + EXPECT_CALL(*core_, Stop(false)).WillOnce(Invoke([&](bool) { + stopped = true; + return Status::OK(); + })); + auto operation = std::async(std::launch::async, [&] { + Attr attr; + return session_->GetAttr(Context{0, 0, 0, 0}, 42, &attr); + }); + auto join = MakeScopedCleanup([&] { + sink->pause.Release(); + if (operation.valid()) operation.wait(); + }); + ASSERT_TRUE(sink->pause.entered.Wait()); + EXPECT_TRUE(business_returned.load()); + EXPECT_EQ(ActiveOperations(), active + 1); + auto stop = std::async(std::launch::async, [&] { return session_->Stop(); }); + auto release_stop = MakeScopedCleanup([&] { sink->pause.Release(); }); + ASSERT_TRUE(hooks.waiting.Wait()); + EXPECT_FALSE(stopped.load()); + sink->pause.Release(); + EXPECT_EQ(operation.get().ToString(), business_status.ToString()); + EXPECT_TRUE(stop.get().ok()); + EXPECT_EQ(ActiveOperations(), active); +} + +TEST_F(ClientSessionLifecycleTest, OwnerJoinsDecrementTailAfterStopReturns) { + OperationHooks hooks(TestPoint::kAfterDecrement); + InstallHooks(hooks); + EXPECT_CALL(*core_, GetInfo(_)).WillOnce(Return(Status::OK())); + EXPECT_CALL(*core_, Stop(false)).WillOnce(Return(Status::OK())); + auto operation = std::async(std::launch::async, [&] { + std::string info; + return session_->GetInfo(&info); + }); + auto join = MakeScopedCleanup([&] { + hooks.pause.Release(); + if (operation.valid()) operation.wait(); + }); + ASSERT_TRUE(hooks.pause.entered.Wait()); + EXPECT_TRUE(session_->Stop().ok()); + EXPECT_EQ(operation.wait_for(kBlockedCheck), std::future_status::timeout); + hooks.pause.Release(); + EXPECT_TRUE(operation.get().ok()); + // Stop's zero-count observation is not permission to destroy until this + // caller has completed its closed-epoch check and notification tail. + session_.reset(); +} + +TEST_F(ClientSessionLifecycleTest, + OwnerJoinsNotifyTailAfterOtherSlotWakesStop) { + OperationHooks hooks(TestPoint::kBeforeNotify); + InstallHooks(hooks); + Pause first_core; + Pause second_core; + std::atomic entered{0}; + EXPECT_CALL(*core_, GetInfo(_)) + .Times(2) + .WillRepeatedly(Invoke([&](std::string*) { + if (entered.fetch_add(1) == 0) { + first_core.Block(); + } else { + second_core.Block(); + } + return Status::OK(); + })); + EXPECT_CALL(*core_, Stop(false)).WillOnce(Return(Status::OK())); + auto first = std::async(std::launch::async, [&] { + OperationTracker::SetTestSlot(0); + auto clear = MakeScopedCleanup([] { OperationTracker::ClearTestSlot(); }); + std::string info; + return session_->GetInfo(&info); + }); + auto join_first = MakeScopedCleanup([&] { + first_core.Release(); + hooks.pause.Release(); + if (first.valid()) first.wait(); + }); + ASSERT_TRUE(first_core.entered.Wait()); + auto second = std::async(std::launch::async, [&] { + OperationTracker::SetTestSlot(1); + auto clear = MakeScopedCleanup([] { OperationTracker::ClearTestSlot(); }); + std::string info; + return session_->GetInfo(&info); + }); + auto join_second = MakeScopedCleanup([&] { + second_core.Release(); + first_core.Release(); + hooks.pause.Release(); + if (second.valid()) second.wait(); + }); + ASSERT_TRUE(second_core.entered.Wait()); + auto stop = std::async(std::launch::async, [&] { return session_->Stop(); }); + auto release_stop = MakeScopedCleanup([&] { + first_core.Release(); + second_core.Release(); + hooks.pause.Release(); + }); + ASSERT_TRUE(hooks.waiting.Wait()); + first_core.Release(); + ASSERT_TRUE(hooks.pause.entered.Wait()); + second_core.Release(); + EXPECT_TRUE(second.get().ok()); + ASSERT_EQ(stop.wait_for(kWaitLimit), std::future_status::ready); + EXPECT_TRUE(stop.get().ok()); + EXPECT_EQ(first.wait_for(kBlockedCheck), std::future_status::timeout); + hooks.pause.Release(); + EXPECT_TRUE(first.get().ok()); + session_.reset(); +} + +TEST_F(ClientSessionLifecycleTest, SuccessfulLocalStartOpensAdmission) { + ResetCreatedSession(); + const auto previous_style = ::testing::FLAGS_gtest_death_test_style; + ::testing::FLAGS_gtest_death_test_style = "threadsafe"; + auto restore = MakeScopedCleanup( + [&] { ::testing::FLAGS_gtest_death_test_style = previous_style; }); + // Start registers process-global loggers; reexec keeps this independent of + // test ordering, repeats, and other fixtures' logger ownership. + ASSERT_EXIT( + { + RunLocalStart(false); + std::_Exit(::testing::Test::HasFailure() ? 1 : 0); + }, + ::testing::ExitedWithCode(0), ""); +} + +TEST_F(ClientSessionLifecycleTest, + StopDuringStartingWaitsForSuccessfulLocalStart) { + ResetCreatedSession(); + const auto previous_style = ::testing::FLAGS_gtest_death_test_style; + ::testing::FLAGS_gtest_death_test_style = "threadsafe"; + auto restore = MakeScopedCleanup( + [&] { ::testing::FLAGS_gtest_death_test_style = previous_style; }); + ASSERT_EXIT( + { + RunLocalStart(true); + std::_Exit(::testing::Test::HasFailure() ? 1 : 0); + }, + ::testing::ExitedWithCode(0), ""); +} + +TEST_F(ClientSessionLifecycleTest, MigratedBthreadReturnsItsOriginalSlot) { + struct Migration { + Pause occupied_worker; + Event operation_done; + bthread_t blocker{}; + int blocker_start{-1}; + pid_t entered_tid{-1}; + pid_t occupied_tid{-1}; + pid_t resumed_tid{-1}; + Status result; + } migration; + + const auto active = ActiveOperations(); + EXPECT_CALL(*core_, GetInfo(_)).WillOnce(Invoke([&](std::string*) { + EXPECT_EQ(ActiveOperations(), active + 1); + // Urgent start transfers this worker to the child. The child blocks the + // OS thread (not just its bthread) until the owner below joins this call. + // Only another worker can steal and resume the public operation. + migration.blocker_start = bthread_start_urgent( + &migration.blocker, nullptr, + +[](void* arg) -> void* { + auto& state = *static_cast(arg); + state.occupied_tid = static_cast(syscall(SYS_gettid)); + state.occupied_worker.Block(); + OperationTracker::ClearTestSlot(); + return nullptr; + }, + &migration); + if (migration.blocker_start != 0) { + return Status::Internal("could not occupy bthread worker"); + } + migration.resumed_tid = static_cast(syscall(SYS_gettid)); + // The exit worker deliberately uses a different slot. Looking up TLS in + // Leave instead of remembering the entry slot would underflow or leak. + OperationTracker::SetTestSlot(1); + return Status::OK(); + })); + EXPECT_CALL(*core_, Stop(false)).WillOnce(Return(Status::OK())); + + std::function call = [&] { + OperationTracker::SetTestSlot(0); + migration.entered_tid = static_cast(syscall(SYS_gettid)); + std::string info; + migration.result = session_->GetInfo(&info); + OperationTracker::ClearTestSlot(); + migration.operation_done.Signal(); + }; + bthread_t caller; + ASSERT_EQ(bthread_start_background( + &caller, nullptr, + +[](void* arg) -> void* { + (*static_cast*>(arg))(); + return nullptr; + }, + &call), + 0); + auto join = MakeScopedCleanup([&] { + migration.occupied_worker.Release(); + bthread_join(caller, nullptr); + if (migration.blocker_start == 0) { + bthread_join(migration.blocker, nullptr); + } + }); + ASSERT_TRUE(migration.operation_done.Wait()); + ASSERT_TRUE(migration.occupied_worker.entered.Wait()); + EXPECT_EQ(migration.blocker_start, 0); + EXPECT_EQ(migration.entered_tid, migration.occupied_tid); + EXPECT_NE(migration.entered_tid, migration.resumed_tid); + RecordProperty("entry_os_tid", migration.entered_tid); + RecordProperty("exit_os_tid", migration.resumed_tid); + EXPECT_TRUE(migration.result.ok()) << migration.result.ToString(); + EXPECT_EQ(ActiveOperations(), active); + EXPECT_TRUE(session_->Stop().ok()); + // Caller and urgent child both join while the session is still owned. +} + +#endif // NDEBUG + } // namespace client } // namespace dingofs diff --git a/test/unit/client/vfs/test_operation_tracker.cc b/test/unit/client/vfs/test_operation_tracker.cc new file mode 100644 index 000000000..c527fff77 --- /dev/null +++ b/test/unit/client/vfs/test_operation_tracker.cc @@ -0,0 +1,684 @@ +/* + * Copyright (c) 2026 dingodb.com, Inc. All Rights Reserved + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include "client/vfs/operation_tracker.h" + +#ifdef NDEBUG +#error "Tracker tests require a Debug build, like the existing SyncPoint tests" +#endif + +namespace dingofs { +namespace client { + +// Test-only access for exact CV handshakes and +// impossible-through-the-public-API invariant failures. Normal admission, +// release, and drain always use real code. +class OperationTrackerTestPeer { + public: + static std::unique_lock LockDrain(OperationTracker& tracker) { + return std::unique_lock(tracker.drain_mutex_); + } + + static void Notify(OperationTracker& tracker) { + tracker.drain_cv_.notify_all(); + } + + static void SetCount(OperationTracker& tracker, unsigned slot, + uint64_t count) { + tracker.counts_[slot].value.store(count, std::memory_order_seq_cst); + } + + static void ReturnSlot(OperationTracker& tracker, unsigned slot) { + tracker.Leave(slot); + } +}; + +namespace { + +using Tracker = OperationTracker; +enum class Point { + kAfterFirstRead, + kAfterIncrement, + kAfterAdmission, + kBeforeWait, + kAfterDecrement, + kBeforeNotify, + kBeforeLeaseRelease, +}; +constexpr auto kCheckpointTimeout = std::chrono::seconds(5); + +// A missing wakeup cannot be repaired safely by destroying a tracker or +// detaching its callers. Contain each concurrency scenario in a death-test +// subprocess: success joins every thread; a bounded synchronization failure +// terminates the entire child, never leaving a live caller in the parent. The +// 20s watchdog is shorter than the production 30s diagnostic wakeup, so that +// wakeup cannot hide a missed notification. No timed negative observation +// proves a CV ordering. +void RunBounded(void (*scenario)()) { + // Other suites may already have background threads; re-exec instead of + // inheriting their potentially locked mutexes into a fork-only child. + const auto previous_style = ::testing::FLAGS_gtest_death_test_style; + ::testing::FLAGS_gtest_death_test_style = "threadsafe"; + EXPECT_EXIT( + { + alarm(20); + scenario(); + if (::testing::Test::HasFailure()) { + std::fprintf(stderr, "Tracker scenario assertion failed\n"); + _exit(1); + } + _exit(0); + }, + ::testing::ExitedWithCode(0), ""); + ::testing::FLAGS_gtest_death_test_style = previous_style; +} + +class Checkpoint { + public: + void Pause() { + std::lock_guard lock(mutex_); + paused_ = true; + } + + void Hit() { + std::unique_lock lock(mutex_); + ++hits_; + cv_.notify_all(); + CHECK(cv_.wait_for(lock, kCheckpointTimeout, [this] { return !paused_; })) + << "Tracker checkpoint was not released"; + } + + void Wait(unsigned count = 1) { + std::unique_lock lock(mutex_); + CHECK(cv_.wait_for(lock, kCheckpointTimeout, + [this, count] { return hits_ >= count; })) + << "Tracker checkpoint not reached: expected " << count << ", got " + << hits_; + } + + void Resume() { + std::lock_guard lock(mutex_); + paused_ = false; + cv_.notify_all(); + } + + unsigned Hits() { + std::lock_guard lock(mutex_); + return hits_; + } + + private: + std::mutex mutex_; + std::condition_variable cv_; + unsigned hits_{0}; + bool paused_{false}; +}; + +class Hooks { + public: + Checkpoint& At(Point point) { return points_[static_cast(point)]; } + + void Attach(Tracker* tracker) { + static const std::array names = { + "OperationTracker::AfterFirstRead", + "OperationTracker::AfterIncrement", + "OperationTracker::AfterAdmission", + "OperationTracker::BeforeWait", + "OperationTracker::AfterDecrement", + "OperationTracker::BeforeNotify", + "OperationTracker::BeforeLeaseRelease", + }; + auto* sync = SyncPoint::GetInstance(); + for (unsigned i = 0; i < names.size(); ++i) { + sync->SetCallBack(names[i], [this, tracker, i](void* argument) { + if (argument == tracker) points_[i].Hit(); + }); + } + sync->EnableProcessing(); + } + + ~Hooks() { + auto* sync = SyncPoint::GetInstance(); + sync->DisableProcessing(); + sync->ClearAllCallBacks(); + } + + private: + std::array points_; +}; + +class ForcedSlot { + public: + explicit ForcedSlot(unsigned slot) { Tracker::SetTestSlot(slot); } + ~ForcedSlot() { Tracker::ClearTestSlot(); } + ForcedSlot(const ForcedSlot&) = delete; + ForcedSlot& operator=(const ForcedSlot&) = delete; +}; + +// Call only after a kBeforeWait observation, with at least one lease still +// held. Acquiring this mutex proves the waiter crossed the CV's atomic unlock; +// a marker outside WaitForDrain could not establish that boundary. +void AcknowledgeWait(Tracker& tracker, Hooks& hooks) { + hooks.At(Point::kBeforeWait).Wait(); + auto lock = OperationTrackerTestPeer::LockDrain(tracker); +} + +// Request a fresh predicate scan while a known lease is still held. The hit +// count is sampled under the drain mutex, so a subsequent hit cannot be an old +// scan. Spurious wakeups are harmless: either way a fresh nonzero predicate +// must be observed, rather than a timeout being used to claim non-completion. +void RequireAnotherWait(Tracker& tracker, Hooks& hooks) { + unsigned next; + { + auto lock = OperationTrackerTestPeer::LockDrain(tracker); + next = hooks.At(Point::kBeforeWait).Hits() + 1; + OperationTrackerTestPeer::Notify(tracker); + } + hooks.At(Point::kBeforeWait).Wait(next); +} + +void RunSlotContention(bool collide) { + Tracker tracker; + Hooks hooks; + hooks.Attach(&tracker); + tracker.OpenOnce(); + std::array admitted; + std::array release; + std::array returned; + std::vector workers; + workers.reserve(Tracker::kSlotCount); + for (unsigned i = 0; i < Tracker::kSlotCount; ++i) { + workers.emplace_back([&, i] { + ForcedSlot slot(collide ? 0 : i); + auto lease = tracker.TryEnter(); + CHECK(lease.has_value()); + admitted[i].Hit(); + release[i].Wait(); + lease.reset(); + returned[i].Hit(); + }); + } + for (auto& checkpoint : admitted) checkpoint.Wait(); + tracker.Close(); + Checkpoint drained; + std::thread waiter([&] { + tracker.WaitForDrain(); + drained.Hit(); + }); + AcknowledgeWait(tracker, hooks); + for (unsigned i = 0; i < Tracker::kSlotCount; ++i) { + release[i].Hit(); + returned[i].Wait(); + if (i + 1 < Tracker::kSlotCount) { + RequireAnotherWait(tracker, hooks); + EXPECT_EQ(drained.Hits(), 0); + } + } + drained.Wait(); + for (auto& worker : workers) worker.join(); + waiter.join(); +} + +TEST(OperationTrackerTest, G01RejectsBeforeOpen) { + Tracker tracker; + unsigned body_calls = 0; + if (auto lease = tracker.TryEnter()) ++body_calls; + EXPECT_EQ(body_calls, 0); + EXPECT_FALSE(tracker.TryEnter().has_value()); +} + +TEST(OperationTrackerTest, G02OpenPublishesInitialization) { + RunBounded([] { + Tracker tracker; + std::array payload{}; + Checkpoint reader_ready; + std::thread reader([&] { + reader_ready.Hit(); + const auto deadline = + std::chrono::steady_clock::now() + kCheckpointTimeout; + for (;;) { + if (auto lease = tracker.TryEnter()) { + EXPECT_EQ(payload[0], 11); + EXPECT_EQ(payload[1], 22); + EXPECT_EQ(payload[2], 33); + EXPECT_EQ(payload[3], 44); + break; + } + CHECK(std::chrono::steady_clock::now() < deadline); + std::this_thread::yield(); + } + }); + reader_ready.Wait(); + // No test synchronization publishes these writes to the reader. Only the + // OpenOnce epoch publication can make them visible before the body runs. + payload = {11, 22, 33, 44}; + tracker.OpenOnce(); + reader.join(); + tracker.Close(); + tracker.WaitForDrain(); + }); +} + +TEST(OperationTrackerDeathTest, G03RepeatedOpenFails) { + EXPECT_DEATH( + { + Tracker tracker; + tracker.OpenOnce(); + tracker.OpenOnce(); + }, + "cannot reopen"); +} + +TEST(OperationTrackerDeathTest, G03ReopenAfterCloseFails) { + EXPECT_DEATH( + { + Tracker tracker; + tracker.OpenOnce(); + tracker.Close(); + tracker.OpenOnce(); + }, + "cannot reopen"); +} + +TEST(OperationTrackerTest, G04RejectsAllEntrantsAfterClose) { + RunBounded([] { + Tracker tracker; + tracker.OpenOnce(); + tracker.Close(); + std::atomic body_calls{0}; + std::vector entrants; + for (unsigned i = 0; i < 16; ++i) { + entrants.emplace_back([&] { + for (unsigned attempt = 0; attempt < 64; ++attempt) { + if (auto lease = tracker.TryEnter()) ++body_calls; + } + }); + } + for (auto& entrant : entrants) entrant.join(); + EXPECT_EQ(body_calls.load(), 0); + tracker.WaitForDrain(); + }); +} + +TEST(OperationTrackerTest, G05EmptyClosedTrackerDrainsWithoutNotification) { + RunBounded([] { + Tracker tracker; + tracker.OpenOnce(); + tracker.Close(); + tracker.WaitForDrain(); + EXPECT_FALSE(tracker.TryEnter().has_value()); + }); +} + +TEST(OperationTrackerTest, G06WaitsForEveryHeldLease) { + RunBounded([] { + Tracker tracker; + Hooks hooks; + hooks.Attach(&tracker); + tracker.OpenOnce(); + std::array, 3> leases; + for (auto& lease : leases) { + lease = tracker.TryEnter(); + CHECK(lease.has_value()); + } + tracker.Close(); + Checkpoint drained; + std::thread waiter([&] { + tracker.WaitForDrain(); + drained.Hit(); + }); + AcknowledgeWait(tracker, hooks); + for (unsigned i = 0; i < 2; ++i) { + leases[i].reset(); + RequireAnotherWait(tracker, hooks); + EXPECT_EQ(drained.Hits(), 0); + } + leases[2].reset(); + drained.Wait(); + waiter.join(); + }); +} + +TEST(OperationTrackerTest, G07MoveConstructionTransfersOneRelease) { + RunBounded([] { + Tracker tracker; + Hooks hooks; + hooks.Attach(&tracker); + tracker.OpenOnce(); + std::optional destination; + { + auto source = tracker.TryEnter(); + CHECK(source.has_value()); + destination.emplace(std::move(*source)); + EXPECT_FALSE(static_cast(*source)); + EXPECT_TRUE(static_cast(*destination)); + } + tracker.Close(); + Checkpoint drained; + std::thread waiter([&] { + tracker.WaitForDrain(); + drained.Hit(); + }); + AcknowledgeWait(tracker, hooks); + EXPECT_EQ(drained.Hits(), 0); + destination.reset(); + drained.Wait(); + waiter.join(); + }); +} + +TEST(OperationTrackerTest, G08MoveAssignmentAcrossTrackersReleasesOldLease) { + RunBounded([] { + Tracker first; + Tracker second; + Hooks hooks; + hooks.Attach(&second); + first.OpenOnce(); + second.OpenOnce(); + auto destination = first.TryEnter(); + auto source = second.TryEnter(); + CHECK(destination.has_value()); + CHECK(source.has_value()); + *destination = std::move(*source); + EXPECT_FALSE(static_cast(*source)); + EXPECT_TRUE(static_cast(*destination)); + source.reset(); + first.Close(); + first.WaitForDrain(); + second.Close(); + Checkpoint drained; + std::thread waiter([&] { + second.WaitForDrain(); + drained.Hit(); + }); + AcknowledgeWait(second, hooks); + EXPECT_EQ(drained.Hits(), 0); + destination.reset(); + drained.Wait(); + waiter.join(); + }); +} + +TEST(OperationTrackerTest, G09ReturnOnAnotherThreadKeepsOriginalSlot) { + RunBounded([] { + Tracker tracker; + Hooks hooks; + hooks.Attach(&tracker); + tracker.OpenOnce(); + std::promise> transferred; + auto incoming = transferred.get_future(); + std::thread entrant([&] { + ForcedSlot slot(7); + transferred.set_value(tracker.TryEnter()); + }); + CHECK(incoming.wait_for(kCheckpointTimeout) == std::future_status::ready); + auto lease = incoming.get(); + CHECK(lease.has_value()); + entrant.join(); + ForcedSlot retained_slot(31); + auto retained = tracker.TryEnter(); + CHECK(retained.has_value()); + tracker.Close(); + std::thread returner([lease = std::move(lease)]() mutable { + ForcedSlot different_slot(31); + lease.reset(); + }); + returner.join(); + Checkpoint drained; + std::thread waiter([&] { + tracker.WaitForDrain(); + drained.Hit(); + }); + AcknowledgeWait(tracker, hooks); + EXPECT_EQ(drained.Hits(), 0); + retained.reset(); + drained.Wait(); + waiter.join(); + }); +} + +TEST(OperationTrackerTest, G10SixtyFourThreadsCollideInOneSlot) { + static_assert(Tracker::kSlotCount == 64, + "Production tracker must stay fixed64"); + RunBounded([] { RunSlotContention(true); }); +} + +TEST(OperationTrackerTest, G11LateTentativeAfterDrainRejectsWithoutBody) { + RunBounded([] { + Tracker tracker; + Hooks hooks; + hooks.At(Point::kAfterFirstRead).Pause(); + hooks.Attach(&tracker); + tracker.OpenOnce(); + unsigned body_calls = 0; + std::thread entrant([&] { + if (auto lease = tracker.TryEnter()) ++body_calls; + }); + hooks.At(Point::kAfterFirstRead).Wait(); + tracker.Close(); + tracker.WaitForDrain(); + // Runtime teardown would now be allowed, but the owner still keeps the + // tracker alive until this rejected caller's rollback and notify tail exit. + hooks.At(Point::kAfterFirstRead).Resume(); + entrant.join(); + EXPECT_EQ(body_calls, 0); + EXPECT_EQ(hooks.At(Point::kAfterIncrement).Hits(), 1); + EXPECT_EQ(hooks.At(Point::kAfterDecrement).Hits(), 1); + }); +} + +TEST(OperationTrackerTest, G12TentativeRollbackWakesDrainer) { + RunBounded([] { + Tracker tracker; + Hooks hooks; + hooks.At(Point::kAfterIncrement).Pause(); + hooks.Attach(&tracker); + tracker.OpenOnce(); + unsigned body_calls = 0; + std::thread entrant([&] { + if (auto lease = tracker.TryEnter()) ++body_calls; + }); + hooks.At(Point::kAfterIncrement).Wait(); + tracker.Close(); + Checkpoint drained; + std::thread waiter([&] { + tracker.WaitForDrain(); + drained.Hit(); + }); + AcknowledgeWait(tracker, hooks); + hooks.At(Point::kAfterIncrement).Resume(); + drained.Wait(); + entrant.join(); + waiter.join(); + EXPECT_EQ(body_calls, 0); + }); +} + +TEST(OperationTrackerTest, G13FinalReleaseBeforePredicateNeedsNoNotification) { + RunBounded([] { + Tracker tracker; + Hooks hooks; + hooks.At(Point::kBeforeNotify).Pause(); + hooks.Attach(&tracker); + tracker.OpenOnce(); + auto lease = tracker.TryEnter(); + CHECK(lease.has_value()); + tracker.Close(); + std::thread returner( + [lease = std::move(lease)]() mutable { lease.reset(); }); + hooks.At(Point::kBeforeNotify).Wait(); + // Counter is already zero; notification is still blocked. Scanning must + // finish without entering CV wait or needing the delayed notification. + tracker.WaitForDrain(); + EXPECT_EQ(hooks.At(Point::kBeforeWait).Hits(), 0); + hooks.At(Point::kBeforeNotify).Resume(); + returner.join(); + }); +} + +TEST(OperationTrackerTest, G14FinalReleaseBetweenPredicateAndWaitIsNotLost) { + RunBounded([] { + Tracker tracker; + Hooks hooks; + hooks.At(Point::kBeforeWait).Pause(); + hooks.Attach(&tracker); + tracker.OpenOnce(); + auto lease = tracker.TryEnter(); + CHECK(lease.has_value()); + tracker.Close(); + Checkpoint drained; + std::thread waiter([&] { + tracker.WaitForDrain(); + drained.Hit(); + }); + hooks.At(Point::kBeforeWait).Wait(); + // The waiter has seen nonzero and STILL HOLDS drain_mutex_. Decrement is + // now forced before the CV atomically drops that mutex and enters wait. + std::thread returner( + [lease = std::move(lease)]() mutable { lease.reset(); }); + hooks.At(Point::kBeforeNotify).Wait(); + EXPECT_EQ(drained.Hits(), 0); + hooks.At(Point::kBeforeWait).Resume(); + drained.Wait(); + returner.join(); + waiter.join(); + }); +} + +TEST(OperationTrackerTest, G15FinalReleaseAfterWaitWakesDrainer) { + RunBounded([] { + Tracker tracker; + Hooks hooks; + hooks.Attach(&tracker); + tracker.OpenOnce(); + auto lease = tracker.TryEnter(); + CHECK(lease.has_value()); + tracker.Close(); + Checkpoint drained; + std::thread waiter([&] { + tracker.WaitForDrain(); + drained.Hit(); + }); + // Tracker mutex acquisition acknowledges actual CV unlock, not merely entry + // into WaitForDrain. The last release only happens after that handshake. + AcknowledgeWait(tracker, hooks); + EXPECT_EQ(drained.Hits(), 0); + lease.reset(); + drained.Wait(); + waiter.join(); + }); +} + +TEST(OperationTrackerTest, G16TentativeRollbackCannotCancelTwoAdmittedLeases) { + RunBounded([] { + Tracker tracker; + Hooks hooks; + hooks.Attach(&tracker); + tracker.OpenOnce(); + ForcedSlot slot(0); + auto first = tracker.TryEnter(); + auto second = tracker.TryEnter(); + CHECK(first.has_value()); + CHECK(second.has_value()); + hooks.At(Point::kAfterIncrement).Pause(); + unsigned body_calls = 0; + std::thread tentative([&] { + ForcedSlot collision(0); + if (auto lease = tracker.TryEnter()) ++body_calls; + }); + hooks.At(Point::kAfterIncrement).Wait(3); + tracker.Close(); + Checkpoint drained; + std::thread waiter([&] { + tracker.WaitForDrain(); + drained.Hit(); + }); + AcknowledgeWait(tracker, hooks); + hooks.At(Point::kAfterIncrement).Resume(); + tentative.join(); + EXPECT_EQ(body_calls, 0); + RequireAnotherWait(tracker, hooks); + EXPECT_EQ(drained.Hits(), 0); + first.reset(); + RequireAnotherWait(tracker, hooks); + EXPECT_EQ(drained.Hits(), 0); + second.reset(); + drained.Wait(); + waiter.join(); + }); +} + +TEST(OperationTrackerTest, G17DrainScansEverySlot) { + RunBounded([] { RunSlotContention(false); }); +} + +TEST(OperationTrackerDeathTest, G18CounterOverflowFails) { + EXPECT_DEATH( + { + Tracker tracker; + ForcedSlot slot(0); + tracker.OpenOnce(); + OperationTrackerTestPeer::SetCount( + tracker, 0, std::numeric_limits::max()); + tracker.TryEnter(); + }, + "counter overflow"); +} + +TEST(OperationTrackerDeathTest, G18CounterUnderflowFails) { + EXPECT_DEATH( + { + Tracker tracker; + OperationTrackerTestPeer::ReturnSlot(tracker, 0); + }, + "underflow or double release"); +} + +TEST(OperationTrackerDeathTest, G18DuplicateReturnFails) { + EXPECT_DEATH(([] { + Tracker tracker; + ForcedSlot slot(0); + tracker.OpenOnce(); + auto lease = tracker.TryEnter(); + CHECK(lease.has_value()); + lease.reset(); + // Simulate a duplicated ownership token without invoking a C++ + // object destructor twice (which would itself be undefined + // behavior). + OperationTrackerTestPeer::ReturnSlot(tracker, 0); + }()), + "underflow or double release"); +} + +} // namespace +} // namespace client +} // namespace dingofs