diff --git a/pulsar-client-cpp/.gitignore b/pulsar-client-cpp/.gitignore index 0d8d323e7ff97..2b3f63fbc75d3 100644 --- a/pulsar-client-cpp/.gitignore +++ b/pulsar-client-cpp/.gitignore @@ -75,6 +75,7 @@ Makefile cmake_install.cmake CMakeFiles CMakeCache.txt +.cmake pulsar-dist install_manifest.txt diff --git a/pulsar-client-cpp/include/pulsar/ConsumerConfiguration.h b/pulsar-client-cpp/include/pulsar/ConsumerConfiguration.h index b326ca8fb3151..e927ad7c884de 100644 --- a/pulsar-client-cpp/include/pulsar/ConsumerConfiguration.h +++ b/pulsar-client-cpp/include/pulsar/ConsumerConfiguration.h @@ -499,6 +499,27 @@ class PULSAR_PUBLIC ConsumerConfiguration { */ bool isAutoAckOldestChunkedMessageOnQueueFull() const; + /** + * If this is enabled, consumer receiver queue size is init as a very small value, 1 by default, + * and it will double itself until it reaches the value set by {@link #receiverQueueSize(int)}, if and only if + * 1) User calls receive() and there is no messages in receiver queue. + * 2) The last message we put in the receiver queue took the last space available in receiver queue. + * + * This is disabled by default and currentReceiverQueueSize is init as maxReceiverQueueSize. + * + * The feature should be able to reduce client memory usage. + * + * Default: false + * + * @param enabled whether to enable AutoScaledReceiverQueueSize. + */ + ConsumerConfiguration& setAutoScaledReceiverQueueSizeEnabled(bool enabled); + + /** + * The associated getter of autoScaledReceiverQueueSizeEnabled + */ + bool isAutoScaledReceiverQueueSizeEnabled() const; + friend class PulsarWrapper; private: diff --git a/pulsar-client-cpp/lib/ClientImpl.cc b/pulsar-client-cpp/lib/ClientImpl.cc index d15e247347a6f..4b504c25a8c85 100644 --- a/pulsar-client-cpp/lib/ClientImpl.cc +++ b/pulsar-client-cpp/lib/ClientImpl.cc @@ -92,7 +92,12 @@ ClientImpl::ClientImpl(const std::string& serviceUrl, const ClientConfiguration& state_(Open), serviceUrl_(serviceUrl), clientConfiguration_(detectTls(serviceUrl, clientConfiguration)), - memoryLimitController_(clientConfiguration.getMemoryLimit()), + memoryLimitController_( + clientConfiguration_.getMemoryLimit(), + clientConfiguration_.getMemoryLimit() * MEMORY_THRESHOLD_FOR_RECEIVER_QUEUE_SIZE_EXPANSION5, + [this]() { + std::for_each(consumers_.begin(), consumers_.end(), [](ConsumerImplBaseWeakPtr consumer) { consumer.lock()->reduceCurrentReceiverQueueSize(); }); + }), ioExecutorProvider_(std::make_shared(clientConfiguration_.getIOThreads())), listenerExecutorProvider_( std::make_shared(clientConfiguration_.getMessageListenerThreads())), diff --git a/pulsar-client-cpp/lib/ConsumerConfiguration.cc b/pulsar-client-cpp/lib/ConsumerConfiguration.cc index 2b58835cdbea3..8e9549539c0be 100644 --- a/pulsar-client-cpp/lib/ConsumerConfiguration.cc +++ b/pulsar-client-cpp/lib/ConsumerConfiguration.cc @@ -19,6 +19,7 @@ #include #include +#include "pulsar/ConsumerConfiguration.h" namespace pulsar { @@ -259,5 +260,12 @@ ConsumerConfiguration& ConsumerConfiguration::setAutoAckOldestChunkedMessageOnQu bool ConsumerConfiguration::isAutoAckOldestChunkedMessageOnQueueFull() const { return impl_->autoAckOldestChunkedMessageOnQueueFull; } +ConsumerConfiguration& ConsumerConfiguration::setAutoScaledReceiverQueueSizeEnabled(bool enabled) { + impl_->autoScaledReceiverQueueSizeEnabled = enabled; + return *this; +} +bool ConsumerConfiguration::isAutoScaledReceiverQueueSizeEnabled() const { + return impl_->autoScaledReceiverQueueSizeEnabled; +} } // namespace pulsar diff --git a/pulsar-client-cpp/lib/ConsumerConfigurationImpl.h b/pulsar-client-cpp/lib/ConsumerConfigurationImpl.h index 1c13f729b55e0..720fe535bb3dc 100644 --- a/pulsar-client-cpp/lib/ConsumerConfigurationImpl.h +++ b/pulsar-client-cpp/lib/ConsumerConfigurationImpl.h @@ -53,6 +53,7 @@ struct ConsumerConfigurationImpl { KeySharedPolicy keySharedPolicy; size_t maxPendingChunkedMessage{10}; bool autoAckOldestChunkedMessageOnQueueFull{false}; + bool autoScaledReceiverQueueSizeEnabled{false}; }; } // namespace pulsar #endif /* LIB_CONSUMERCONFIGURATIONIMPL_H_ */ diff --git a/pulsar-client-cpp/lib/ConsumerImpl.cc b/pulsar-client-cpp/lib/ConsumerImpl.cc index 63d2afc29bcab..57b0b9df95f77 100644 --- a/pulsar-client-cpp/lib/ConsumerImpl.cc +++ b/pulsar-client-cpp/lib/ConsumerImpl.cc @@ -31,11 +31,14 @@ #include "AckGroupingTrackerDisabled.h" #include #include +#include namespace pulsar { DECLARE_LOG_OBJECT() +const static int INITIAL_RECEIVER_QUEUE_SIZE = 1; + ConsumerImpl::ConsumerImpl(const ClientImplPtr client, const std::string& topic, const std::string& subscriptionName, const ConsumerConfiguration& conf, const ExecutorServicePtr listenerExecutor /* = NULL by default */, @@ -55,6 +58,7 @@ ConsumerImpl::ConsumerImpl(const ClientImplPtr client, const std::string& topic, // This is the initial capacity of the queue incomingMessages_(std::max(config_.getReceiverQueueSize(), 1)), availablePermits_(0), + scaleReceiverQueueHint(false), receiverQueueRefillThreshold_(config_.getReceiverQueueSize() / 2), consumerId_(client->newConsumerId()), consumerName_(config_.getConsumerName()), @@ -103,6 +107,7 @@ ConsumerImpl::ConsumerImpl(const ClientImplPtr client, const std::string& topic, if (conf.isEncryptionEnabled()) { msgCrypto_ = std::make_shared(consumerStr_, false); } + initReceiverQueueSize(); } ConsumerImpl::~ConsumerImpl() { @@ -880,10 +885,12 @@ Optional ConsumerImpl::clearReceiveQueue() { void ConsumerImpl::increaseAvailablePermits(const ClientConnectionPtr& currentCnx, int delta) { int newAvailablePermits = availablePermits_.fetch_add(delta) + delta; - while (newAvailablePermits >= receiverQueueRefillThreshold_ && messageListenerRunning_) { + while (newAvailablePermits >= getCurrentReceiverQueueSize() / 2 && messageListenerRunning_) { if (availablePermits_.compare_exchange_weak(newAvailablePermits, 0)) { sendFlowPermitsToBroker(currentCnx, newAvailablePermits); break; + } else { + newAvailablePermits = availablePermits_; } } } @@ -1385,4 +1392,51 @@ bool ConsumerImpl::isConnected() const { uint64_t ConsumerImpl::getNumberOfConnectedConsumer() { return isConnected() ? 1 : 0; } -} /* namespace pulsar */ +MemoryLimitController& ConsumerImpl::getMemoryLimitController() { + return client_.lock()->getMemoryLimitController(); +} + +void ConsumerImpl::reduceCurrentReceiverQueueSize() { + if (!config_.isAutoScaledReceiverQueueSizeEnabled()) { + return ; + } + int oldSize = getCurrentReceiverQueueSize(); + int newSize = std::max(minReceiverQueueSize(), oldSize / 2); + if (oldSize > newSize) { + setCurrentReceiverQueueSize(newSize); + } +} + +void ConsumerImpl::expectMoreIncomingMessages() { + if (!config_.isAutoScaledReceiverQueueSizeEnabled()) { + return ; + } + double usage = getMemoryLimitController().currentUsagePercent(); + if (bool expectedState = true && usage < MEMORY_THRESHOLD_FOR_RECEIVER_QUEUE_SIZE_EXPANSION5 + && scaleReceiverQueueHint.compare_exchange_strong(expectedState, false)) { + int oldSize = getCurrentReceiverQueueSize(); + int newSize = std::min(config_.getReceiverQueueSize(), oldSize * 2); + setCurrentReceiverQueueSize(newSize); + } +} +void ConsumerImpl::initReceiverQueueSize() { + if (config_.isAutoScaledReceiverQueueSizeEnabled()) { + int size = minReceiverQueueSize(); + currentReceiverQueueSize_.exchange(size); + } else { + currentReceiverQueueSize_.exchange(config_.getReceiverQueueSize()); + } +} + +void ConsumerImpl::setCurrentReceiverQueueSize(int newSize) { + currentReceiverQueueSize_.fetch_xor(newSize); +} + +int ConsumerImpl::getCurrentReceiverQueueSize() { + return currentReceiverQueueSize_; +} +int ConsumerImpl::minReceiverQueueSize() { + int size = std::min(INITIAL_RECEIVER_QUEUE_SIZE, config_.getReceiverQueueSize()); + return size; +} +} /* namespace pulsar */ \ No newline at end of file diff --git a/pulsar-client-cpp/lib/ConsumerImpl.h b/pulsar-client-cpp/lib/ConsumerImpl.h index 346d3515ad7f6..ea561ea8f7705 100644 --- a/pulsar-client-cpp/lib/ConsumerImpl.h +++ b/pulsar-client-cpp/lib/ConsumerImpl.h @@ -129,6 +129,7 @@ class ConsumerImpl : public ConsumerImplBase, void negativeAcknowledge(const MessageId& msgId) override; bool isConnected() const override; uint64_t getNumberOfConnectedConsumer() override; + void reduceCurrentReceiverQueueSize() override; virtual void disconnectConsumer(); Result fetchSingleMessageFromBroker(Message& msg); @@ -141,6 +142,7 @@ class ConsumerImpl : public ConsumerImplBase, virtual bool isReadCompacted(); virtual void hasMessageAvailableAsync(HasMessageAvailableCallback callback); virtual void getLastMessageIdAsync(BrokerGetLastMessageIdCallback callback); + int getCurrentReceiverQueueSize(); protected: // overrided methods from HandlerBase @@ -157,6 +159,11 @@ class ConsumerImpl : public ConsumerImplBase, void handleClose(Result result, ResultCallback callback, ConsumerImplPtr consumer); ConsumerStatsBasePtr consumerStatsBasePtr_; + void setCurrentReceiverQueueSize(int newSize); + void expectMoreIncomingMessages(); + void initReceiverQueueSize(); + int minReceiverQueueSize(); + private: bool waitingForZeroQueueSizeMessage; bool uncompressMessageIfNeeded(const ClientConnectionPtr& cnx, const proto::MessageIdData& messageIdData, @@ -173,6 +180,8 @@ class ConsumerImpl : public ConsumerImplBase, bool decryptMessageIfNeeded(const ClientConnectionPtr& cnx, const proto::CommandMessage& msg, const proto::MessageMetadata& metadata, SharedBuffer& payload); + MemoryLimitController& getMemoryLimitController(); + // TODO - Convert these functions to lambda when we move to C++11 Result receiveHelper(Message& msg); Result receiveHelper(Message& msg, int timeout); @@ -199,6 +208,8 @@ class ConsumerImpl : public ConsumerImplBase, UnboundedBlockingQueue incomingMessages_; std::queue pendingReceives_; std::atomic_int availablePermits_; + std::atomic_int currentReceiverQueueSize_; + std::atomic_bool scaleReceiverQueueHint; const int receiverQueueRefillThreshold_; uint64_t consumerId_; std::string consumerName_; diff --git a/pulsar-client-cpp/lib/ConsumerImplBase.h b/pulsar-client-cpp/lib/ConsumerImplBase.h index 693d4da9a3779..ab250cbfbbd0b 100644 --- a/pulsar-client-cpp/lib/ConsumerImplBase.h +++ b/pulsar-client-cpp/lib/ConsumerImplBase.h @@ -57,6 +57,7 @@ class ConsumerImplBase { virtual void negativeAcknowledge(const MessageId& msgId) = 0; virtual bool isConnected() const = 0; virtual uint64_t getNumberOfConnectedConsumer() = 0; + virtual void reduceCurrentReceiverQueueSize() = 0; private: virtual void setNegativeAcknowledgeEnabledForTesting(bool enabled) = 0; diff --git a/pulsar-client-cpp/lib/MemoryLimitController.cc b/pulsar-client-cpp/lib/MemoryLimitController.cc index f55da8e68a4c5..284d53a975924 100644 --- a/pulsar-client-cpp/lib/MemoryLimitController.cc +++ b/pulsar-client-cpp/lib/MemoryLimitController.cc @@ -22,7 +22,12 @@ namespace pulsar { MemoryLimitController::MemoryLimitController(uint64_t memoryLimit) - : memoryLimit_(memoryLimit), currentUsage_(0), mutex_(), condition_() {} + : memoryLimit_(memoryLimit), currentUsage_(0), mutex_(), condition_(), + triggerThreshold_(0), trigger_(), triggerRunning(false) {} + +MemoryLimitController::MemoryLimitController(uint64_t memoryLimit, uint64_t triggerThreshold, Trigger trigger) + : memoryLimit_(memoryLimit), currentUsage_(0), mutex_(), condition_(), triggerThreshold_(triggerThreshold), + trigger_(trigger), triggerRunning(false) {} bool MemoryLimitController::tryReserveMemory(uint64_t size) { // Avoid CAS operation when size is 0 @@ -40,11 +45,30 @@ bool MemoryLimitController::tryReserveMemory(uint64_t size) { } if (currentUsage_.compare_exchange_strong(current, newUsage)) { + checkTrigger(current, newUsage); return true; } } } +void MemoryLimitController::forceReserveMemory(uint64_t size) { + uint64_t newUsage = currentUsage_.fetch_add(size); + checkTrigger(newUsage - size, newUsage); +} + +void MemoryLimitController::checkTrigger(uint64_t preUsage, uint64_t newUsage) { + if (newUsage >= triggerThreshold_ && preUsage < triggerThreshold_ && trigger_) { + bool expectedState = false; + if (triggerRunning.compare_exchange_strong(expectedState, true)) { + try { + trigger_(); + } catch (const std::exception exception) { + } + triggerRunning.exchange(false); + } + } +} + bool MemoryLimitController::reserveMemory(uint64_t size) { if (!tryReserveMemory(size)) { std::unique_lock lock(mutex_); @@ -83,4 +107,8 @@ void MemoryLimitController::close() { condition_.notify_all(); } +double MemoryLimitController::currentUsagePercent() const { + return 1.0 * currentUsage_ / memoryLimit_; +} + } // namespace pulsar diff --git a/pulsar-client-cpp/lib/MemoryLimitController.h b/pulsar-client-cpp/lib/MemoryLimitController.h index 38987ea0b68a0..23df2cbfdf2d2 100644 --- a/pulsar-client-cpp/lib/MemoryLimitController.h +++ b/pulsar-client-cpp/lib/MemoryLimitController.h @@ -25,14 +25,20 @@ #include namespace pulsar { +typedef std::function Trigger; + +const static double MEMORY_THRESHOLD_FOR_RECEIVER_QUEUE_SIZE_EXPANSION5 = 0.75; class MemoryLimitController { public: explicit MemoryLimitController(uint64_t memoryLimit); + MemoryLimitController(uint64_t memoryLimit, uint64_t triggerThreshold, Trigger trigger); + void forceReserveMemory(uint64_t size); bool tryReserveMemory(uint64_t size); bool reserveMemory(uint64_t size); void releaseMemory(uint64_t size); uint64_t currentUsage() const; + double currentUsagePercent() const; void close(); @@ -42,6 +48,10 @@ class MemoryLimitController { std::mutex mutex_; std::condition_variable condition_; bool isClosed_ = false; + const uint64_t triggerThreshold_; + const Trigger trigger_; + std::atomic_bool triggerRunning; + void checkTrigger(uint64_t preUsage, uint64_t newUsage); }; } // namespace pulsar \ No newline at end of file diff --git a/pulsar-client-cpp/lib/MultiTopicsConsumerImpl.cc b/pulsar-client-cpp/lib/MultiTopicsConsumerImpl.cc index 0ad9d60681048..c8a00e020f9b4 100644 --- a/pulsar-client-cpp/lib/MultiTopicsConsumerImpl.cc +++ b/pulsar-client-cpp/lib/MultiTopicsConsumerImpl.cc @@ -738,3 +738,4 @@ uint64_t MultiTopicsConsumerImpl::getNumberOfConnectedConsumer() { }); return numberOfConnectedConsumer; } +void MultiTopicsConsumerImpl::reduceCurrentReceiverQueueSize() {} diff --git a/pulsar-client-cpp/lib/MultiTopicsConsumerImpl.h b/pulsar-client-cpp/lib/MultiTopicsConsumerImpl.h index 98b2f318af95a..c0755624ac1c5 100644 --- a/pulsar-client-cpp/lib/MultiTopicsConsumerImpl.h +++ b/pulsar-client-cpp/lib/MultiTopicsConsumerImpl.h @@ -80,6 +80,7 @@ class MultiTopicsConsumerImpl : public ConsumerImplBase, void negativeAcknowledge(const MessageId& msgId) override; bool isConnected() const override; uint64_t getNumberOfConnectedConsumer() override; + void reduceCurrentReceiverQueueSize() override; void handleGetConsumerStats(Result, BrokerConsumerStats, LatchPtr, MultiTopicsBrokerConsumerStatsPtr, size_t, BrokerConsumerStatsCallback); diff --git a/pulsar-client-cpp/lib/PartitionedConsumerImpl.cc b/pulsar-client-cpp/lib/PartitionedConsumerImpl.cc index 23288e2f81108..f4b3ef2c038b7 100644 --- a/pulsar-client-cpp/lib/PartitionedConsumerImpl.cc +++ b/pulsar-client-cpp/lib/PartitionedConsumerImpl.cc @@ -642,5 +642,6 @@ uint64_t PartitionedConsumerImpl::getNumberOfConnectedConsumer() { } return numberOfConnectedConsumer; } +void PartitionedConsumerImpl::reduceCurrentReceiverQueueSize() {} } // namespace pulsar diff --git a/pulsar-client-cpp/lib/PartitionedConsumerImpl.h b/pulsar-client-cpp/lib/PartitionedConsumerImpl.h index 7fa0ccdd1f4c9..7a822178e737d 100644 --- a/pulsar-client-cpp/lib/PartitionedConsumerImpl.h +++ b/pulsar-client-cpp/lib/PartitionedConsumerImpl.h @@ -75,6 +75,7 @@ class PartitionedConsumerImpl : public ConsumerImplBase, void negativeAcknowledge(const MessageId& msgId) override; bool isConnected() const override; uint64_t getNumberOfConnectedConsumer() override; + void reduceCurrentReceiverQueueSize() override; void handleGetConsumerStats(Result, BrokerConsumerStats, LatchPtr, PartitionedBrokerConsumerStatsPtr, size_t, BrokerConsumerStatsCallback); diff --git a/pulsar-client-cpp/tests/MemoryLimitControllerTest.cc b/pulsar-client-cpp/tests/MemoryLimitControllerTest.cc index eb63760eedfba..b00cac6eb00b1 100644 --- a/pulsar-client-cpp/tests/MemoryLimitControllerTest.cc +++ b/pulsar-client-cpp/tests/MemoryLimitControllerTest.cc @@ -41,6 +41,31 @@ TEST(MemoryLimitControllerTest, testLimit) { ASSERT_EQ(mlc.currentUsage(), 101); } +TEST(MemoryLimitControllerTest, testTrigger) { + int num = 0; + MemoryLimitController mlc(100, 95, [&num]() { + num++; + }); + + mlc.tryReserveMemory(95); + ASSERT_EQ(num , 1); + + mlc.releaseMemory(95); + ASSERT_EQ(mlc.currentUsage(), 0); + + std::thread t1([&]() { + mlc.forceReserveMemory(95); + }); + + std::thread t2([&]() { + mlc.forceReserveMemory(95); + }); + + t1.join(); + t2.join(); + ASSERT_EQ(num, 2); +} + TEST(MemoryLimitControllerTest, testBlocking) { MemoryLimitController mlc(100);