fix(imc): make frame timestamp cache thread-local and fix has_pending race

Convert g_FrameTimestampNs from a global variable to thread_local storage
to prevent timestamp leakage across threads in multi-threaded publish
scenarios.

Use atomic exchange for has_pending flag in Pump() to eliminate the race
between checking and clearing. Set has_pending after successful enqueue
in both backpressure paths and re-check queue after drain to handle
concurrent publishes correctly.

Add test to verify frame timestamp cache isolation between threads.

Author: doyaGu

src/Core/ImcBusPublish.cpp

@@ -212,6 +212,7 @@ namespace BML::Core {
         case BML_BACKPRESSURE_DROP_OLDEST:
             ApplyBackpressure(sub, message);
             if (sub->queue && sub->queue->Enqueue(message, message->priority)) {
+                sub->has_pending.store(true, std::memory_order_release);
                 sub->RecordReceived(message->Size());
                 m_GlobalStats.total_messages_delivered.fetch_add(
                     1, std::memory_order_relaxed);
@@ -230,6 +231,7 @@ namespace BML::Core {
             for (int spin = 0; spin < 100; ++spin) {
                 std::this_thread::yield();
                 if (sub->queue && sub->queue->Enqueue(message, message->priority)) {
+                    sub->has_pending.store(true, std::memory_order_release);
                     sub->RecordReceived(message->Size());
                     m_GlobalStats.total_messages_delivered.fetch_add(
                         1, std::memory_order_relaxed);
@@ -915,30 +917,33 @@ namespace BML::Core {
 
     void ImcBusImpl::Pump(size_t max_per_sub) {
         // Cache timestamp for the entire frame to avoid repeated QPC syscalls.
-        g_FrameTimestampNs = GetTimestampNsRaw();
+        SetFrameTimestampCache(GetTimestampNsRaw());
 
         m_GlobalStats.pump_cycles.fetch_add(1, std::memory_order_relaxed);
-        m_GlobalStats.last_pump_time.store(g_FrameTimestampNs, std::memory_order_relaxed);
+        m_GlobalStats.last_pump_time.store(GetFrameTimestampCache(),
+                                           std::memory_order_relaxed);
 
         DrainRpcQueue(max_per_sub);
 
         {
             SnapshotGuard guard(m_PublishState.snapshot.load(std::memory_order_acquire));
             if (!guard) {
-                g_FrameTimestampNs = 0;
+                SetFrameTimestampCache(0);
                 return;
             }
 
             for (auto *sub : guard.get()->all_subs) {
                 if (sub->closed.load(std::memory_order_acquire))
                     continue;
                 // Fast skip: no pending messages means empty queue, avoid drain overhead.
-                if (!sub->has_pending.load(std::memory_order_acquire))
+                if (!sub->has_pending.exchange(false, std::memory_order_acq_rel))
                     continue;
-                sub->has_pending.store(false, std::memory_order_relaxed);
                 sub->ref_count.fetch_add(1, std::memory_order_relaxed);
                 DrainSubscription(sub, max_per_sub);
                 sub->ref_count.fetch_sub(1, std::memory_order_acq_rel);
+                if (sub->queue && !sub->queue->IsEmpty()) {
+                    sub->has_pending.store(true, std::memory_order_release);
+                }
             }
         }
 
@@ -952,6 +957,6 @@ namespace BML::Core {
             }
         }
 
-        g_FrameTimestampNs = 0;
+        SetFrameTimestampCache(0);
     }
 } // namespace BML::Core

src/Core/ImcBusSharedInternal.h

@@ -62,13 +62,20 @@ namespace BML::Core {
             duration_cast<nanoseconds>(steady_clock::now().time_since_epoch()).count());
     }
 
-    // Cached per-frame timestamp. Set once at Pump() start, used for all
-    // message timestamps and stats within the same frame. Eliminates
-    // repeated QPC syscalls (~20-50ns each, ~12+/frame).
-    inline uint64_t g_FrameTimestampNs = 0;
+    // Cached per-frame timestamp for the current thread. Set once at Pump() start,
+    // used for all same-thread publishes and stats within that pump frame.
+    inline uint64_t &GetFrameTimestampCache() noexcept {
+        static thread_local uint64_t frameTimestampNs = 0;
+        return frameTimestampNs;
+    }
+
+    inline void SetFrameTimestampCache(uint64_t timestamp) noexcept {
+        GetFrameTimestampCache() = timestamp;
+    }
 
     inline uint64_t GetTimestampNs() noexcept {
-        return g_FrameTimestampNs ? g_FrameTimestampNs : GetTimestampNsRaw();
+        const uint64_t frameTimestampNs = GetFrameTimestampCache();
+        return frameTimestampNs ? frameTimestampNs : GetTimestampNsRaw();
     }
 
     template <typename T, size_t N>

tests/ImcBusTest.cpp

@@ -20,6 +20,7 @@
 
 #include "Core/Context.h"
 #include "Core/ImcBus.h"
+#include "Core/ImcBusSharedInternal.h"
 #include "TestKernel.h"
 #include "TestKernelBuilder.h"
 #include "TestModHelper.h"
@@ -1137,6 +1138,13 @@ TEST_F(ImcBusTest, PumpBudgetDistributesLoadAcrossSubscribers) {
     EXPECT_EQ(cap1.count.load(), 1);
     EXPECT_EQ(cap2.count.load(), 1);
 
+    BML_SubscriptionStats stats1 = BML_SUBSCRIPTION_STATS_INIT;
+    BML_SubscriptionStats stats2 = BML_SUBSCRIPTION_STATS_INIT;
+    ASSERT_EQ(BML_RESULT_OK, ImcGetSubscriptionStats(sub1, &stats1));
+    ASSERT_EQ(BML_RESULT_OK, ImcGetSubscriptionStats(sub2, &stats2));
+    EXPECT_EQ(stats1.queue_size, 3u);
+    EXPECT_EQ(stats2.queue_size, 3u);
+
     // Drain remaining
     Pump(0);
     EXPECT_EQ(cap1.count.load(), 4);
@@ -1146,6 +1154,33 @@ TEST_F(ImcBusTest, PumpBudgetDistributesLoadAcrossSubscribers) {
     EXPECT_EQ(BML_RESULT_OK, ImcUnsubscribe(sub2));
 }
 
+TEST(ImcBusSharedInternalTest, FrameTimestampCacheDoesNotLeakAcrossThreads) {
+    constexpr uint64_t kCachedTimestamp = 0x123456789ABCDEF0ull;
+
+    std::atomic<bool> workerReady{false};
+    std::atomic<bool> workerRelease{false};
+
+    std::thread worker([&] {
+        BML::Core::SetFrameTimestampCache(kCachedTimestamp);
+        workerReady.store(true, std::memory_order_release);
+        while (!workerRelease.load(std::memory_order_acquire)) {
+            std::this_thread::yield();
+        }
+        BML::Core::SetFrameTimestampCache(0);
+    });
+
+    while (!workerReady.load(std::memory_order_acquire)) {
+        std::this_thread::yield();
+    }
+
+    const uint64_t observedTimestamp = BML::Core::GetTimestampNs();
+
+    workerRelease.store(true, std::memory_order_release);
+    worker.join();
+
+    EXPECT_NE(observedTimestamp, kCachedTimestamp);
+}
+
 TEST_F(ImcBusTest, PublishingResumesAfterPumpClearsQueue) {
     // Backpressure scenario: queue fills up, pump clears, publish succeeds again
     BML_TopicId topic;