perf(common): gate IOHook shared_mutex behind armed flag

Try() is invoked by CHECK_HOOK on every local-file IO, so taking a
std::shared_mutex shared lock per call regresses the previously
lock-free IO path. Writers (Reset()/Clear()) only exist in tests and
the production default is the disabled state, so gate the synchronized
path behind an atomic armed flag: Try() is now a single acquire load
when disabled, and only takes the mutex once armed.

Reset() publishes the complete configuration under the mutex and then
release-stores armed; armed readers acquire the mutex before reading
mode/pos, so an observed armed state always implies a complete
configuration. Clear() disarms first so IO threads drop off the lock
as soon as possible.

One semantic change: the disabled fast path no longer increments the
IO counter, so IOCount() now counts only while armed; all IOCount()
consumers are tests and the header documents the contract. Add
IOHookTest.TestDisabledFastPath to pin the disabled behavior, and arm
the hook in TestReadAheadCache.TestPreBufferWindowLimit (#209), which
relied on the previous count-while-disabled behavior.

Author: u70b3

src/paimon/common/factories/io_hook.cpp

@@ -24,13 +24,50 @@
 #include <stdexcept>
 
 #include "fmt/format.h"
+#include "paimon/macros.h"
 #include "paimon/status.h"
 
 namespace paimon {
 
 class IOHook::Impl {
  public:
     Status Try(const std::string& path) {
+        // Fast path: the hook is disabled, which is always the case in production;
+        // writers (Reset()/Clear()) only exist in tests. This keeps Try() a single
+        // atomic load on the IO path instead of a shared_mutex acquisition per IO.
+        if (PAIMON_UNLIKELY(armed_.load(std::memory_order_acquire))) {
+            return TryArmed(path);
+        }
+        return Status::OK();
+    }
+
+    inline void Reset(int64_t pos, IOHook::Mode mode) {
+        std::unique_lock<std::shared_mutex> lock(mutex_);
+        mode_ = mode;
+        pos_ = pos;
+        io_count_ = 0;
+        // Arm only after the configuration is complete: TryArmed() reads mode_/pos_
+        // under mutex_, which synchronizes with this store, so an observed armed state
+        // always implies a complete configuration.
+        armed_.store(true, std::memory_order_release);
+    }
+
+    int64_t IOCount() const {
+        std::shared_lock<std::shared_mutex> lock(mutex_);
+        return io_count_.load();
+    }
+
+    void Clear() {
+        std::unique_lock<std::shared_mutex> lock(mutex_);
+        // Disarm first so IO threads stop taking the lock as soon as possible.
+        armed_.store(false, std::memory_order_release);
+        mode_ = IOHook::Mode::SILENT;
+        pos_ = -1;
+        io_count_ = 0;
+    }
+
+ private:
+    Status TryArmed(const std::string& path) {
         std::shared_lock<std::shared_mutex> lock(mutex_);
         if (io_count_.fetch_add(1) < pos_) {
             return Status::OK();
@@ -51,24 +88,8 @@ class IOHook::Impl {
         }
     }
 
-    inline void Reset(int64_t pos, IOHook::Mode mode) {
-        std::unique_lock<std::shared_mutex> lock(mutex_);
-        mode_ = mode;
-        pos_ = pos;
-        io_count_ = 0;
-    }
-
-    int64_t IOCount() const {
-        std::shared_lock<std::shared_mutex> lock(mutex_);
-        return io_count_.load();
-    }
-
-    void Clear() {
-        Reset(-1, IOHook::Mode::SILENT);
-    }
-
- private:
     mutable std::shared_mutex mutex_;
+    std::atomic<bool> armed_ = {false};
     std::atomic<int64_t> io_count_ = {0};
     int64_t pos_ = -1;
     IOHook::Mode mode_ = IOHook::Mode::SILENT;

src/paimon/common/factories/io_hook.h

@@ -45,7 +45,8 @@ class PAIMON_EXPORT IOHook : public Singleton<IOHook> {
     };
 
     /// Reset the IO exception position and behavior mode to handle the exception.
-    /// IOCount will be reset to 0.
+    /// IOCount will be reset to 0. Arms the hook: Try() switches from its lock-free
+    /// disabled fast path to the synchronized armed path.
     ///
     /// @params pos The position where the IO exception occurs.
     /// @params mode The mode of behavior for handling the exception.
@@ -56,12 +57,14 @@ class PAIMON_EXPORT IOHook : public Singleton<IOHook> {
     Status Try(const std::string& path);
 
     /// Get the count of IO operations that have already occurred.
+    /// IOs are only counted while the hook is armed (after Reset(), before Clear());
+    /// the disabled fast path does not count.
     ///
     /// @return The number of IO operations executed.
     int64_t IOCount() const;
 
     /// Clear the state of the IOHook, including resetting IO count and
-    /// any stored exception state.
+    /// any stored exception state. Disarms the hook back to the lock-free fast path.
     void Clear();
 
  private:

src/paimon/common/factories/io_hook_test.cpp

@@ -68,12 +68,33 @@ TEST(IOHookTest, TestThrowExceptionMode) {
     hook->Clear();
 }
 
+// The disabled state is the production default: Try() must take the lock-free fast
+// path, always return OK, and not count IOs (see IOCount()'s contract). Clear() first
+// so the test does not depend on execution order.
+TEST(IOHookTest, TestDisabledFastPath) {
+    auto hook = IOHook::GetInstance();
+    hook->Clear();
+    ASSERT_OK(hook->Try("path"));
+    ASSERT_OK(hook->Try("path"));
+    ASSERT_EQ(0, hook->IOCount());
+
+    // Re-arming and disarming must restore the exact disabled behavior.
+    hook->Reset(0, IOHook::Mode::RETURN_ERROR);
+    ASSERT_NOK(hook->Try("path"));
+    ASSERT_EQ(1, hook->IOCount());
+    hook->Clear();
+    ASSERT_OK(hook->Try("path"));
+    ASSERT_OK(hook->Try("path"));
+    ASSERT_EQ(0, hook->IOCount());
+}
+
 // Regression test for torn IOHook configurations: Reset()/Clear() run on one thread
 // while other threads call Try() concurrently. A shared start barrier releases all
 // threads together, and the reset thread keeps hammering until every worker has
-// finished, so overlap is structural rather than timing-dependent. Under a
-// ThreadSanitizer build this deterministically reports any unsynchronized access;
-// functionally every Try() must return OK.
+// finished, so overlap is structural rather than timing-dependent. The continuous
+// arm/disarm cycling also keeps workers switching between the disabled fast path and
+// the synchronized armed path. Under a ThreadSanitizer build this deterministically
+// reports any unsynchronized access; functionally every Try() must return OK.
 TEST(IOHookTest, TestConcurrentResetAndTry) {
     auto hook = IOHook::GetInstance();
 

src/paimon/common/utils/read_ahead_cache_test.cpp

@@ -464,7 +464,8 @@ TEST(TestReadAheadCache, TestPreBufferWindowLimit) {
 
     auto io_hook = paimon::IOHook::GetInstance();
     paimon::ScopeGuard guard([&io_hook]() { io_hook->Clear(); });
-    io_hook->Clear();
+    // IOCount() only counts while armed; INT64_MAX never triggers the error mode.
+    io_hook->Reset(INT64_MAX, paimon::IOHook::Mode::RETURN_ERROR);
 
     AssertReadEquals({0, 10}, "abcdefghij", &cache);
     // The second range did not fit into the window: only one prefetch IO.
@@ -475,7 +476,7 @@ TEST(TestReadAheadCache, TestPreBufferWindowLimit) {
     ASSERT_EQ(io_hook->IOCount(), 2);
 
     // The range is cached now: re-reading it issues no IO at all.
-    io_hook->Clear();
+    io_hook->Reset(INT64_MAX, paimon::IOHook::Mode::RETURN_ERROR);
     AssertReadEquals({16, 10}, "qrstuvwxyz", &cache);
     ASSERT_EQ(io_hook->IOCount(), 0);
 }