Hash thread id by index

Author: BewareMyPower

lib/ClientImpl.cc

@@ -516,7 +516,7 @@ void ClientImpl::handleConsumerCreated(Result result, ConsumerImplBaseWeakPtr co
     }
 }
 
-Future<Result, ClientConnectionPtr> ClientImpl::getConnection(const std::string& topic, int key) {
+Future<Result, ClientConnectionPtr> ClientImpl::getConnection(const std::string& topic, size_t key) {
     Promise<Result, ClientConnectionPtr> promise;
 
     const auto topicNamePtr = TopicName::get(topic);

lib/ClientImpl.h

@@ -95,7 +95,7 @@ class ClientImpl : public std::enable_shared_from_this<ClientImpl> {
 
     void getPartitionsForTopicAsync(const std::string& topic, GetPartitionsCallback callback);
 
-    Future<Result, ClientConnectionPtr> getConnection(const std::string& topic, int key);
+    Future<Result, ClientConnectionPtr> getConnection(const std::string& topic, size_t key);
 
     void closeAsync(CloseCallback callback);
     void shutdown();

lib/ConnectionPool.cc

@@ -63,7 +63,7 @@ bool ConnectionPool::close() {
 
 Future<Result, ClientConnectionWeakPtr> ConnectionPool::getConnectionAsync(const std::string& logicalAddress,
                                                                            const std::string& physicalAddress,
-                                                                           int keySuffix) {
+                                                                           size_t keySuffix) {
     if (closed_) {
         Promise<Result, ClientConnectionWeakPtr> promise;
         promise.setFailed(ResultAlreadyClosed);
@@ -96,8 +96,7 @@ Future<Result, ClientConnectionWeakPtr> ConnectionPool::getConnectionAsync(const
     // No valid or pending connection found in the pool, creating a new one
     ClientConnectionPtr cnx;
     try {
-        size_t index = executorIndex_++;
-        cnx.reset(new ClientConnection(logicalAddress, physicalAddress, executorProvider_->get(index),
+        cnx.reset(new ClientConnection(logicalAddress, physicalAddress, executorProvider_->get(keySuffix),
                                        clientConfiguration_, authentication_, clientVersion_, *this));
     } catch (const std::runtime_error& e) {
         lock.unlock();

lib/ConnectionPool.h

@@ -75,7 +75,7 @@ class PULSAR_PUBLIC ConnectionPool {
      */
     Future<Result, ClientConnectionWeakPtr> getConnectionAsync(const std::string& logicalAddress,
                                                                const std::string& physicalAddress,
-                                                               int keySuffix);
+                                                               size_t keySuffix);
 
     Future<Result, ClientConnectionWeakPtr> getConnectionAsync(const std::string& logicalAddress,
                                                                const std::string& physicalAddress) {
@@ -86,7 +86,7 @@ class PULSAR_PUBLIC ConnectionPool {
         return getConnectionAsync(address, address);
     }
 
-    int generateRandomIndex() { return randomDistribution_(randomEngine_); }
+    size_t generateRandomIndex() { return randomDistribution_(randomEngine_); }
 
    private:
     ClientConfiguration clientConfiguration_;
@@ -97,8 +97,6 @@ class PULSAR_PUBLIC ConnectionPool {
     const std::string clientVersion_;
     mutable std::recursive_mutex mutex_;
     std::atomic_bool closed_{false};
-    // Use a separated index so that connections will be distributed uniformly across the executors
-    std::atomic_size_t executorIndex_{0};
 
     std::uniform_int_distribution<> randomDistribution_;
     std::mt19937 randomEngine_;

lib/HandlerBase.h

@@ -99,7 +99,7 @@ class HandlerBase : public std::enable_shared_from_this<HandlerBase> {
 
    protected:
     ClientImplWeakPtr client_;
-    const int connectionKeySuffix_;
+    const size_t connectionKeySuffix_;
     ExecutorServicePtr executor_;
     mutable std::mutex mutex_;
     std::mutex pendingReceiveMutex_;