From 9d1902341a6855ccfdb38646ca4458ce45e743f0 Mon Sep 17 00:00:00 2001 From: Alexey Milovidov Date: Fri, 16 Oct 2026 20:42:26 +0000 Subject: [PATCH] Make the random name and connection index generators thread-safe `generateRandomName` mutated a process-wide `std::mt19937` without synchronization, so subscriptions created concurrently by different clients raced on it (undefined behavior, reported by TSan). Use thread-local generator state instead. `ConnectionPool::generateRandomIndex` is called from both the caller and the executor threads, so guard its generator with a mutex. Co-Authored-By: Claude Opus 5.5 --- lib/ClientImpl.cc | 7 ++++--- lib/ConnectionPool.h | 7 ++++++- 2 files changed, 10 insertions(+), 4 deletions(-) diff --git a/lib/ClientImpl.cc b/lib/ClientImpl.cc index 043d6d42..acac66c2 100644 --- a/lib/ClientImpl.cc +++ b/lib/ClientImpl.cc @@ -78,11 +78,12 @@ void invokeLegacyCallback(const std::function& callback, static const char hexDigits[] = {'0', '1', '2', '3', '4', '5', '6', '7', '8', '9', 'a', 'b', 'c', 'd', 'e', 'f'}; -static std::uniform_int_distribution<> hexDigitsDist(0, sizeof(hexDigits) - 1); -static std::mt19937 randomEngine = - std::mt19937(std::chrono::high_resolution_clock::now().time_since_epoch().count()); std::string generateRandomName() { + // Called concurrently by all the clients of the process, so the generator state must not be shared. + thread_local std::uniform_int_distribution<> hexDigitsDist(0, sizeof(hexDigits) - 1); + thread_local std::mt19937 randomEngine(std::random_device{}()); + const int randomNameLength = 10; std::string randomName; diff --git a/lib/ConnectionPool.h b/lib/ConnectionPool.h index 201c1d01..8acfa329 100644 --- a/lib/ConnectionPool.h +++ b/lib/ConnectionPool.h @@ -96,7 +96,11 @@ class PULSAR_PUBLIC ConnectionPool { return getConnectionAsync(address, address); } - size_t generateRandomIndex() { return randomDistribution_(randomEngine_); } + size_t generateRandomIndex() { + // Handlers are created concurrently from the caller and executor threads. + std::lock_guard lock(randomMutex_); + return randomDistribution_(randomEngine_); + } private: const AtomicSharedPtr& serviceInfo_; @@ -110,6 +114,7 @@ class PULSAR_PUBLIC ConnectionPool { std::uniform_int_distribution<> randomDistribution_; std::mt19937 randomEngine_; + std::mutex randomMutex_; auto releaseConnections() { decltype(pool_) pool;