Replace ClientSession admission mutex with OperationTracker

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);

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.
         """

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();

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
 )
 

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::OperationLease>
 ClientSession::TryAcquireOperation() {
-  {
-    std::lock_guard<std::mutex> 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<std::mutex> 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<std::mutex> 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);
 

src/client/vfs/client_session.h

@@ -26,6 +26,7 @@
 #include <utility>
 #include <vector>
 
+#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<OperationLease> 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};

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 <chrono>
+
+namespace dingofs {
+namespace client {
+namespace {
+
+std::atomic<uint64_t> 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<uint64_t>::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<std::mutex> 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