From a4e199fdd492a06fc94cc12679a4ecdb7f6664c9 Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Thu, 16 Dec 2021 12:52:43 +0800 Subject: [PATCH 01/27] Support configuring chunkingEnabled --- .../include/pulsar/ProducerConfiguration.h | 18 ++++++++++++++++++ pulsar-client-cpp/lib/ProducerConfiguration.cc | 7 +++++++ .../lib/ProducerConfigurationImpl.h | 1 + .../tests/ProducerConfigurationTest.cc | 4 ++++ 4 files changed, 30 insertions(+) diff --git a/pulsar-client-cpp/include/pulsar/ProducerConfiguration.h b/pulsar-client-cpp/include/pulsar/ProducerConfiguration.h index 5c2792aadeaa3..1fbaa1dd4892c 100644 --- a/pulsar-client-cpp/include/pulsar/ProducerConfiguration.h +++ b/pulsar-client-cpp/include/pulsar/ProducerConfiguration.h @@ -473,6 +473,24 @@ class PULSAR_PUBLIC ProducerConfiguration { */ ProducerConfiguration& setProperties(const std::map& properties); + /** + * If message size is higher than allowed max publish-payload size by broker then enableChunking helps + * producer to split message into multiple chunks and publish them to broker separately in order. So, it + * allows client to successfully publish large size of messages in pulsar. + * + * Set it true to enable this feature. + * + * Default: false + * + * @return the ProducerConfiguration self + */ + ProducerConfiguration& setChunkingEnabled(bool chunkingEnabled); + + /** + * The getter associated with setChunkingEnabled(). + */ + bool isChunkingEnabled() const; + friend class PulsarWrapper; private: diff --git a/pulsar-client-cpp/lib/ProducerConfiguration.cc b/pulsar-client-cpp/lib/ProducerConfiguration.cc index 3e027eeb1975f..0ee38ca92e025 100644 --- a/pulsar-client-cpp/lib/ProducerConfiguration.cc +++ b/pulsar-client-cpp/lib/ProducerConfiguration.cc @@ -251,4 +251,11 @@ ProducerConfiguration& ProducerConfiguration::setProperties( return *this; } +ProducerConfiguration& ProducerConfiguration::setChunkingEnabled(bool chunkingEnabled) { + impl_->chunkingEnabled = chunkingEnabled; + return *this; +} + +bool ProducerConfiguration::isChunkingEnabled() const { return impl_->chunkingEnabled; } + } // namespace pulsar diff --git a/pulsar-client-cpp/lib/ProducerConfigurationImpl.h b/pulsar-client-cpp/lib/ProducerConfigurationImpl.h index a41b2507ea43c..2ac1ebaa5df1a 100644 --- a/pulsar-client-cpp/lib/ProducerConfigurationImpl.h +++ b/pulsar-client-cpp/lib/ProducerConfigurationImpl.h @@ -48,6 +48,7 @@ struct ProducerConfigurationImpl { std::set encryptionKeys; ProducerCryptoFailureAction cryptoFailureAction{ProducerCryptoFailureAction::FAIL}; std::map properties; + bool chunkingEnabled{false}; }; } // namespace pulsar diff --git a/pulsar-client-cpp/tests/ProducerConfigurationTest.cc b/pulsar-client-cpp/tests/ProducerConfigurationTest.cc index b88f6e41890d8..5c541295ff966 100644 --- a/pulsar-client-cpp/tests/ProducerConfigurationTest.cc +++ b/pulsar-client-cpp/tests/ProducerConfigurationTest.cc @@ -46,6 +46,7 @@ TEST(ProducerConfigurationTest, testDefaultConfig) { ASSERT_EQ(conf.isEncryptionEnabled(), false); ASSERT_EQ(conf.getEncryptionKeys(), std::set{}); ASSERT_EQ(conf.getProperties().empty(), true); + ASSERT_EQ(conf.isChunkingEnabled(), false); } class MockMessageRoutingPolicy : public MessageRoutingPolicy { @@ -129,4 +130,7 @@ TEST(ProducerConfigurationTest, testCustomConfig) { conf.setProperty("k1", "v1"); ASSERT_EQ(conf.getProperties()["k1"], "v1"); ASSERT_EQ(conf.hasProperty("k1"), true); + + conf.setChunkingEnabled(true); + ASSERT_EQ(conf.isChunkingEnabled(), true); } From dfd069b4ada714267dc75965916dda3d3a533291 Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Thu, 16 Dec 2021 14:57:31 +0800 Subject: [PATCH 02/27] Add limitations when chunking is enabled --- .../include/pulsar/ProducerConfiguration.h | 12 +++++++++++- pulsar-client-cpp/lib/ClientImpl.cc | 4 ++++ 2 files changed, 15 insertions(+), 1 deletion(-) diff --git a/pulsar-client-cpp/include/pulsar/ProducerConfiguration.h b/pulsar-client-cpp/include/pulsar/ProducerConfiguration.h index 1fbaa1dd4892c..7c278dd6e9124 100644 --- a/pulsar-client-cpp/include/pulsar/ProducerConfiguration.h +++ b/pulsar-client-cpp/include/pulsar/ProducerConfiguration.h @@ -478,10 +478,20 @@ class PULSAR_PUBLIC ProducerConfiguration { * producer to split message into multiple chunks and publish them to broker separately in order. So, it * allows client to successfully publish large size of messages in pulsar. * - * Set it true to enable this feature. + * Set it true to enable this feature. If so, you must disable batching (see setBatchingEnabled), + * otherwise the producer creation will fail. + * + * There are some other recommendations when it's enabled: + * 1. This features is right now only supported for non-shared subscription and persistent-topic. + * 2. It's better to reduce setMaxPendingMessages to avoid producer accupying large amount of memory by + * buffered messages. + * 3. Set message-ttl on the namespace to cleanup chunked messages. Sometimes due to broker-restart or + * publish time, producer might fail to publish entire large message. So, consumer will not be able to + * consume and ack those messages. * * Default: false * + * @param chunkingEnabled whether chunking is enabled * @return the ProducerConfiguration self */ ProducerConfiguration& setChunkingEnabled(bool chunkingEnabled); diff --git a/pulsar-client-cpp/lib/ClientImpl.cc b/pulsar-client-cpp/lib/ClientImpl.cc index 8cdaacc8b2523..9f1e904bdf448 100644 --- a/pulsar-client-cpp/lib/ClientImpl.cc +++ b/pulsar-client-cpp/lib/ClientImpl.cc @@ -29,6 +29,7 @@ #include #include #include +#include #include #include #include @@ -147,6 +148,9 @@ LookupServicePtr ClientImpl::getLookup() { return lookupServicePtr_; } void ClientImpl::createProducerAsync(const std::string& topic, ProducerConfiguration conf, CreateProducerCallback callback) { + if (conf.isChunkingEnabled() && conf.getBatchingEnabled()) { + throw std::invalid_argument("Batching and chunking of messages can't be enabled together"); + } TopicNamePtr topicName; { Lock lock(mutex_); From 97b72348475be7f2f3310ecb09e0f13f3550ab4c Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Mon, 20 Dec 2021 13:17:33 +0800 Subject: [PATCH 03/27] [WIP] serializeAndSendMessage --- pulsar-client-cpp/lib/ProducerImpl.cc | 105 ++++++++++++++++++------ pulsar-client-cpp/lib/ProducerImpl.h | 25 ++++++ pulsar-client-cpp/tests/ProducerTest.cc | 11 ++- 3 files changed, 113 insertions(+), 28 deletions(-) diff --git a/pulsar-client-cpp/lib/ProducerImpl.cc b/pulsar-client-cpp/lib/ProducerImpl.cc index f81e205475de4..62b07c0f77433 100644 --- a/pulsar-client-cpp/lib/ProducerImpl.cc +++ b/pulsar-client-cpp/lib/ProducerImpl.cc @@ -57,7 +57,8 @@ ProducerImpl::ProducerImpl(ClientImplPtr client, const std::string& topic, const producerId_(client->newProducerId()), msgSequenceGenerator_(0), dataKeyGenIntervalSec_(4 * 60 * 60), - memoryLimitController_(client->getMemoryLimitController()) { + memoryLimitController_(client->getMemoryLimitController()), + chunkingEnabled_(conf_.isChunkingEnabled() && !conf_.getBatchingEnabled()) { LOG_DEBUG("ProducerName - " << producerName_ << " Created producer on topic " << topic_ << " id: " << producerId_); @@ -369,10 +370,12 @@ void ProducerImpl::sendAsync(const Message& msg, SendCallback callback) { uint32_t uncompressedSize = payload.readableBytes(); uint32_t payloadSize = uncompressedSize; + bool compressed = false; ClientConnectionPtr cnx = getCnx().lock(); - if (!batchMessageContainer_) { + if (!batchMessageContainer_ || msg.impl_->metadata.has_deliver_at_time()) { // If batching is enabled we compress all the payloads together before sending the batch payload = CompressionCodecProvider::getCodec(conf_.getCompressionType()).encode(payload); + compressed = true; payloadSize = payload.readableBytes(); // Encrypt the payload if enabled @@ -383,7 +386,7 @@ void ProducerImpl::sendAsync(const Message& msg, SendCallback callback) { } payload = encryptedPayload; - if (payloadSize > ClientConnection::getMaxMessageSize()) { + if (payloadSize > ClientConnection::getMaxMessageSize() && !chunkingEnabled_) { LOG_DEBUG(getName() << " - compressed Message payload size" << payloadSize << "cannot exceed " << ClientConnection::getMaxMessageSize() << " bytes"); cb(ResultMessageTooBig, msg.getMessageId()); @@ -391,21 +394,29 @@ void ProducerImpl::sendAsync(const Message& msg, SendCallback callback) { } } + const auto messageId = msg.getMessageId(); + if (msg.impl_->metadata.has_producer_name()) { + // Message had already been sent before + releaseSemaphore(payloadSize); + cb(ResultInvalidMessage, messageId); + return; + } + + const int totalChunks = + conf_.getBatchingEnabled() + ? 1 + : getNumOfChunks(uncompressedSize, static_cast(ClientConnection::getMaxMessageSize())); + // Each chunk should be sent individually, so try to acquire extra permits for chunks. + for (int i = 0; i < totalChunks - 1; i++) { + if (!canEnqueueRequest(0, messageId, cb)) { + return; + } + } + // Reserve a spot in the messages queue before acquiring the ProducerImpl // mutex. When the queue is full, this call will block until a spot is // available. - Result res = canEnqueueRequest(payloadSize); - if (res != ResultOk) { - // If queue is full sending the batch immediately, no point waiting till batchMessagetimeout - if (batchMessageContainer_) { - LOG_DEBUG(getName() << " - sending batch message immediately"); - Lock lock(mutex_); - auto failures = batchMessageAndSend(); - lock.unlock(); - failures.complete(); - } - - cb(res, msg.getMessageId()); + if (!canEnqueueRequest(payloadSize, messageId, cb)) { return; } @@ -414,15 +425,7 @@ void ProducerImpl::sendAsync(const Message& msg, SendCallback callback) { if (state_ != Ready && state_ != Pending) { lock.unlock(); releaseSemaphore(payloadSize); - cb(ResultAlreadyClosed, msg.getMessageId()); - return; - } - - if (msg.impl_->metadata.has_producer_name()) { - // Message had already been sent before - lock.unlock(); - releaseSemaphore(payloadSize); - cb(ResultInvalidMessage, msg.getMessageId()); + cb(ResultAlreadyClosed, messageId); return; } @@ -434,7 +437,29 @@ void ProducerImpl::sendAsync(const Message& msg, SendCallback callback) { } setMessageMetadata(msg, sequenceId, uncompressedSize); - // If we reach this point then you have a reserved spot on the queue + const std::string uuid = (totalChunks > 1) ? (producerName_ + "-" + std::to_string(sequenceId)) : ""; + const std::string schemaVersion = (totalChunks > 1 && msg.impl_->metadata.has_schema_version()) + ? msg.impl_->metadata.schema_version() + : ""; + int readStartIndex = 0; + for (int chunkId = 0; chunkId < totalChunks; chunkId++) { + // The schema version might change after serializeAndSendMessage(), so we need to reset it. + if (chunkId > 0 && msg.impl_->metadata.has_schema_version()) { + msg.impl_->metadata.set_schema_version(schemaVersion); + } + serializeAndSendMessage(msg, msg.impl_->payload, sequenceId, uuid, chunkId, totalChunks, + readStartIndex, ClientConnection::getMaxMessageSize(), payload, compressed, + payloadSize, uncompressedSize, cb); + readStartIndex = (chunkId + 1) * ClientConnection::getMaxMessageSize(); + } +} + +void ProducerImpl::serializeAndSendMessage(const Message& msg, SharedBuffer& payload, uint64_t sequenceId, + const std::string& uuid, int chunkId, int totalChunks, + int readStartIndex, int chunkMaxSizeInBytes, + SharedBuffer& compressedPayload, bool compressed, + int compressedPayloadSize, int uncompressedSize, SendCallback cb) { + // TODO: implement the real logic if (batchMessageContainer_ && !msg.impl_->metadata.has_deliver_at_time()) { // Batching is enabled and the message is not delayed if (!batchMessageContainer_->hasEnoughSpace(msg)) { @@ -451,12 +476,20 @@ void ProducerImpl::sendAsync(const Message& msg, SendCallback callback) { if (isFull) { auto failures = batchMessageAndSend(); - lock.unlock(); + // lock.unlock(); failures.complete(); } } else { - sendMessage(OpSendMsg{msg, cb, producerId_, sequenceId, conf_.getSendTimeout(), 1, payloadSize}); + sendMessage(OpSendMsg{msg, cb, producerId_, sequenceId, conf_.getSendTimeout(), 1, + static_cast(compressedPayloadSize)}); + } +} + +int ProducerImpl::getNumOfChunks(uint32_t size, uint32_t maxMessageSize) { + if (size >= maxMessageSize && maxMessageSize != 0) { + return size / maxMessageSize + ((size % maxMessageSize == 0) ? 0 : 1); } + return 1; } Result ProducerImpl::canEnqueueRequest(uint32_t payloadSize) { @@ -485,6 +518,24 @@ Result ProducerImpl::canEnqueueRequest(uint32_t payloadSize) { } } +bool ProducerImpl::canEnqueueRequest(uint32_t payloadSize, const MessageId& messageId, + const SendCallback& cb) { + const auto result = canEnqueueRequest(payloadSize); + if (result != ResultOk) { + // If queue is full sending the batch immediately, no point waiting till batchMessagetimeout + if (batchMessageContainer_) { + LOG_DEBUG(getName() << " - sending batch message immediately"); + Lock lock(mutex_); + auto failures = batchMessageAndSend(); + lock.unlock(); + failures.complete(); + } + + cb(result, messageId); + } + return result == ResultOk; +} + void ProducerImpl::releaseSemaphore(uint32_t payloadSize) { if (semaphore_) { semaphore_->release(); diff --git a/pulsar-client-cpp/lib/ProducerImpl.h b/pulsar-client-cpp/lib/ProducerImpl.h index d29efed1a13ae..ec84ba407a52f 100644 --- a/pulsar-client-cpp/lib/ProducerImpl.h +++ b/pulsar-client-cpp/lib/ProducerImpl.h @@ -83,6 +83,8 @@ class ProducerImpl : public HandlerBase, int32_t partition() const noexcept { return partition_; } + static int getNumOfChunks(uint32_t size, uint32_t maxMessageSize); + protected: ProducerStatsBasePtr producerStatsBasePtr_; @@ -125,12 +127,34 @@ class ProducerImpl : public HandlerBase, bool encryptMessage(proto::MessageMetadata& metadata, SharedBuffer& payload, SharedBuffer& encryptedPayload); + /** + * Reserve a spot in the messages queue before acquiring the ProducerImpl mutex. When the queue is full, + * this call will block until a spot is available if blockIfQueueIsFull is true. Otherwise, it will return + * ResultProducerQueueIsFull immediately. + * + * It also checks whether the memory could reach the limit after `payloadSize` is added. If so, this call + * will block until enough memory could be retained. + */ Result canEnqueueRequest(uint32_t payloadSize); + + /** + * It calls the previous overloaded method. If the result is not ResultOk, `cb` will be completed with + * `result` and `messageId` and `batchMessageAndSend` will be called to send all pending messages. + */ + bool canEnqueueRequest(uint32_t payloadSize, const MessageId& messageId, const SendCallback& cb); + + void serializeAndSendMessage(const Message& msg, SharedBuffer& payload, uint64_t sequenceId, + const std::string& uuid, int chunkId, int totalChunks, int readStartIndex, + int chunkMaxSizeInBytes, SharedBuffer& compressedPayload, bool compressed, + int compressedPayloadSize, int uncompressedSize, SendCallback cb); + void releaseSemaphore(uint32_t payloadSize); void releaseSemaphoreForSendOp(const OpSendMsg& op); void cancelTimers(); + bool canAddToBatch(const Message& msg) const; + typedef std::unique_lock Lock; ProducerConfiguration conf_; @@ -169,6 +193,7 @@ class ProducerImpl : public HandlerBase, uint32_t dataKeyGenIntervalSec_; MemoryLimitController& memoryLimitController_; + const bool chunkingEnabled_; }; struct ProducerImplCmp { diff --git a/pulsar-client-cpp/tests/ProducerTest.cc b/pulsar-client-cpp/tests/ProducerTest.cc index 210f01345d4af..3ee8def37dbaf 100644 --- a/pulsar-client-cpp/tests/ProducerTest.cc +++ b/pulsar-client-cpp/tests/ProducerTest.cc @@ -26,6 +26,7 @@ #include "lib/Utils.h" #include "lib/Latch.h" #include "lib/LogUtils.h" +#include "lib/ProducerImpl.h" DECLARE_LOG_OBJECT() using namespace pulsar; @@ -240,4 +241,12 @@ TEST(ProducerTest, testSendAsyncCloseAsyncConcurrentlyWithLazyProducers) { client.close(); LOG_INFO("End of run " << run); } -} \ No newline at end of file +} + +TEST(ProducerTest, testGetNumOfChunks) { + ASSERT_EQ(ProducerImpl::getNumOfChunks(11, 5), 3); + ASSERT_EQ(ProducerImpl::getNumOfChunks(10, 5), 2); + ASSERT_EQ(ProducerImpl::getNumOfChunks(8, 5), 2); + ASSERT_EQ(ProducerImpl::getNumOfChunks(4, 5), 1); + ASSERT_EQ(ProducerImpl::getNumOfChunks(1, 0), 1); +} From 69a911b62c0fbb6eb057a28b0abe41506d512927 Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Tue, 21 Dec 2021 18:39:29 +0800 Subject: [PATCH 04/27] Pass TopicName instead of string to ProducerImpl's constructor --- pulsar-client-cpp/lib/ClientImpl.cc | 2 +- pulsar-client-cpp/lib/PartitionedProducerImpl.cc | 5 ++--- pulsar-client-cpp/lib/ProducerImpl.cc | 6 +++--- pulsar-client-cpp/lib/ProducerImpl.h | 3 ++- pulsar-client-cpp/lib/TopicName.cc | 4 ++-- pulsar-client-cpp/lib/TopicName.h | 4 ++-- 6 files changed, 12 insertions(+), 12 deletions(-) diff --git a/pulsar-client-cpp/lib/ClientImpl.cc b/pulsar-client-cpp/lib/ClientImpl.cc index 9f1e904bdf448..0b07b6e9f2d5c 100644 --- a/pulsar-client-cpp/lib/ClientImpl.cc +++ b/pulsar-client-cpp/lib/ClientImpl.cc @@ -178,7 +178,7 @@ void ClientImpl::handleCreateProducer(const Result result, const LookupDataResul producer = std::make_shared(shared_from_this(), topicName, partitionMetadata->getPartitions(), conf); } else { - producer = std::make_shared(shared_from_this(), topicName->toString(), conf); + producer = std::make_shared(shared_from_this(), *topicName, conf); } producer->getProducerCreatedFuture().addListener( std::bind(&ClientImpl::handleProducerCreated, shared_from_this(), std::placeholders::_1, diff --git a/pulsar-client-cpp/lib/PartitionedProducerImpl.cc b/pulsar-client-cpp/lib/PartitionedProducerImpl.cc index bdd23ed6c91c2..4f8be3b92becc 100644 --- a/pulsar-client-cpp/lib/PartitionedProducerImpl.cc +++ b/pulsar-client-cpp/lib/PartitionedProducerImpl.cc @@ -86,8 +86,7 @@ unsigned int PartitionedProducerImpl::getNumPartitionsWithLock() const { ProducerImplPtr PartitionedProducerImpl::newInternalProducer(unsigned int partition, bool lazy) { using namespace std::placeholders; - std::string topicPartitionName = topicName_->getTopicPartitionName(partition); - auto producer = std::make_shared(client_, topicPartitionName, conf_, partition); + auto producer = std::make_shared(client_, *topicName_, conf_, partition); if (lazy) { createLazyPartitionProducer(partition); @@ -97,7 +96,7 @@ ProducerImplPtr PartitionedProducerImpl::newInternalProducer(unsigned int partit const_cast(this)->shared_from_this(), _1, _2, partition)); } - LOG_DEBUG("Creating Producer for single Partition - " << topicPartitionName); + LOG_DEBUG("Creating Producer for single Partition - " << topicName_ << "-partition-" << partition); return producer; } diff --git a/pulsar-client-cpp/lib/ProducerImpl.cc b/pulsar-client-cpp/lib/ProducerImpl.cc index 62b07c0f77433..c383773edfa1b 100644 --- a/pulsar-client-cpp/lib/ProducerImpl.cc +++ b/pulsar-client-cpp/lib/ProducerImpl.cc @@ -42,10 +42,10 @@ struct ProducerImpl::PendingCallbacks { } }; -ProducerImpl::ProducerImpl(ClientImplPtr client, const std::string& topic, const ProducerConfiguration& conf, - int32_t partition) +ProducerImpl::ProducerImpl(ClientImplPtr client, const TopicName& topicName, + const ProducerConfiguration& conf, int32_t partition) : HandlerBase( - client, topic, + client, (partition < 0) ? topicName.toString() : topicName.getTopicPartitionName(partition), Backoff(milliseconds(100), seconds(60), milliseconds(std::max(100, conf.getSendTimeout() - 100)))), conf_(conf), semaphore_(), diff --git a/pulsar-client-cpp/lib/ProducerImpl.h b/pulsar-client-cpp/lib/ProducerImpl.h index ec84ba407a52f..05b81e7b69e28 100644 --- a/pulsar-client-cpp/lib/ProducerImpl.h +++ b/pulsar-client-cpp/lib/ProducerImpl.h @@ -47,12 +47,13 @@ class PulsarFriend; class Producer; class MemoryLimitController; +class TopicName; class ProducerImpl : public HandlerBase, public std::enable_shared_from_this, public ProducerImplBase { public: - ProducerImpl(ClientImplPtr client, const std::string& topic, + ProducerImpl(ClientImplPtr client, const TopicName& topic, const ProducerConfiguration& producerConfiguration, int32_t partition = -1); ~ProducerImpl(); diff --git a/pulsar-client-cpp/lib/TopicName.cc b/pulsar-client-cpp/lib/TopicName.cc index a56b9795f396e..2e6232cf001de 100644 --- a/pulsar-client-cpp/lib/TopicName.cc +++ b/pulsar-client-cpp/lib/TopicName.cc @@ -216,7 +216,7 @@ std::string TopicName::getLookupName() { return ss.str(); } -std::string TopicName::toString() { +std::string TopicName::toString() const { std::stringstream ss; std::string seperator("/"); if (isV2Topic_ && cluster_.empty()) { @@ -230,7 +230,7 @@ std::string TopicName::toString() { bool TopicName::isPersistent() const { return this->domain_ == TopicDomain::Persistent; } -const std::string TopicName::getTopicPartitionName(unsigned int partition) { +std::string TopicName::getTopicPartitionName(unsigned int partition) const { std::stringstream topicPartitionName; // make this topic name as well topicPartitionName << toString() << PartitionedProducerImpl::PARTITION_NAME_SUFFIX << partition; diff --git a/pulsar-client-cpp/lib/TopicName.h b/pulsar-client-cpp/lib/TopicName.h index 248001c172a41..1d5deab553dca 100644 --- a/pulsar-client-cpp/lib/TopicName.h +++ b/pulsar-client-cpp/lib/TopicName.h @@ -55,14 +55,14 @@ class PULSAR_PUBLIC TopicName : public ServiceUnitId { std::string getNamespacePortion(); std::string getLocalName(); std::string getEncodedLocalName(); - std::string toString(); + std::string toString() const; bool isPersistent() const; NamespaceNamePtr getNamespaceName(); int getPartitionIndex() const noexcept { return partition_; } static std::shared_ptr get(const std::string& topicName); bool operator==(const TopicName& other); static std::string getEncodedName(const std::string& nameBeforeEncoding); - const std::string getTopicPartitionName(unsigned int partition); + std::string getTopicPartitionName(unsigned int partition) const; static int getPartitionIndex(const std::string& topic); private: From 4dcec9e6832c608ece9f6b6b9b8dcb07cfa7ecac Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Wed, 22 Dec 2021 17:56:26 +0800 Subject: [PATCH 05/27] Send messages by chunks --- pulsar-client-cpp/lib/OpSendMsg.h | 12 + pulsar-client-cpp/lib/ProducerImpl.cc | 210 ++++++++++-------- pulsar-client-cpp/lib/ProducerImpl.h | 17 +- pulsar-client-cpp/lib/SharedBuffer.h | 4 +- .../lib/stats/ProducerStatsBase.h | 2 +- .../lib/stats/ProducerStatsDisabled.h | 2 +- .../lib/stats/ProducerStatsImpl.cc | 2 +- .../lib/stats/ProducerStatsImpl.h | 2 +- 8 files changed, 136 insertions(+), 115 deletions(-) diff --git a/pulsar-client-cpp/lib/OpSendMsg.h b/pulsar-client-cpp/lib/OpSendMsg.h index 70b880cf5179c..dac04004b8da1 100644 --- a/pulsar-client-cpp/lib/OpSendMsg.h +++ b/pulsar-client-cpp/lib/OpSendMsg.h @@ -36,6 +36,8 @@ struct OpSendMsg { boost::posix_time::ptime timeout_; uint32_t messagesCount_; uint64_t messagesSize_; + int totalChunks_ = 0; + int chunkId_ = -1; OpSendMsg() = default; @@ -48,6 +50,16 @@ struct OpSendMsg { timeout_(TimeUtils::now() + milliseconds(sendTimeoutMs)), messagesCount_(messagesCount), messagesSize_(messagesSize) {} + + OpSendMsg& setTotalChunks(int totalChunks) { + totalChunks_ = totalChunks; + return *this; + } + + OpSendMsg& setChunkId(int chunkId) { + chunkId_ = chunkId; + return *this; + } }; } // namespace pulsar diff --git a/pulsar-client-cpp/lib/ProducerImpl.cc b/pulsar-client-cpp/lib/ProducerImpl.cc index c383773edfa1b..4b59eb5627932 100644 --- a/pulsar-client-cpp/lib/ProducerImpl.cc +++ b/pulsar-client-cpp/lib/ProducerImpl.cc @@ -58,7 +58,7 @@ ProducerImpl::ProducerImpl(ClientImplPtr client, const TopicName& topicName, msgSequenceGenerator_(0), dataKeyGenIntervalSec_(4 * 60 * 60), memoryLimitController_(client->getMemoryLimitController()), - chunkingEnabled_(conf_.isChunkingEnabled() && !conf_.getBatchingEnabled()) { + chunkingEnabled_(conf_.isChunkingEnabled() && topicName.isPersistent() && !conf_.getBatchingEnabled()) { LOG_DEBUG("ProducerName - " << producerName_ << " Created producer on topic " << topic_ << " id: " << producerId_); @@ -324,14 +324,6 @@ void ProducerImpl::setMessageMetadata(const Message& msg, const uint64_t& sequen } } -void ProducerImpl::statsCallBackHandler(Result res, const MessageId& msgId, SendCallback callback, - boost::posix_time::ptime publishTime) { - producerStatsBasePtr_->messageReceived(res, publishTime); - if (callback) { - callback(res, msgId); - } -} - void ProducerImpl::flushAsync(FlushCallback callback) { if (batchMessageContainer_) { Lock lock(mutex_); @@ -359,108 +351,97 @@ void ProducerImpl::triggerFlush() { } } -void ProducerImpl::sendAsync(const Message& msg, SendCallback callback) { - producerStatsBasePtr_->messageSent(msg); - SendCallback cb = - std::bind(&ProducerImpl::statsCallBackHandler, shared_from_this(), std::placeholders::_1, - std::placeholders::_2, callback, boost::posix_time::microsec_clock::universal_time()); +bool ProducerImpl::isValidProducerState(const SendCallback& callback) { + Lock lock(mutex_); + const auto state = state_; + lock.unlock(); + switch (state) { + case HandlerBase::Ready: + // OK + case HandlerBase::Pending: + // We are OK to queue the messages on the client, it will be sent to the broker once we get the + // connectionPL + case HandlerBase::Closing: + case HandlerBase::Closed: + callback(ResultAlreadyClosed, {}); + return false; + case HandlerBase::NotStarted: + case HandlerBase::Failed: + default: + callback(ResultNotConnected, {}); + return false; + } +} - // Compress the payload if required - SharedBuffer& payload = msg.impl_->payload; +bool ProducerImpl::canAddToBatch(const Message& msg) const { + // If a message has a delayed delivery time, we'll always send it individually + return batchMessageContainer_.get() && !msg.impl_->metadata.has_deliver_at_time(); +} - uint32_t uncompressedSize = payload.readableBytes(); - uint32_t payloadSize = uncompressedSize; - bool compressed = false; - ClientConnectionPtr cnx = getCnx().lock(); - if (!batchMessageContainer_ || msg.impl_->metadata.has_deliver_at_time()) { - // If batching is enabled we compress all the payloads together before sending the batch - payload = CompressionCodecProvider::getCodec(conf_.getCompressionType()).encode(payload); - compressed = true; - payloadSize = payload.readableBytes(); - - // Encrypt the payload if enabled - SharedBuffer encryptedPayload; - if (!encryptMessage(msg.impl_->metadata, payload, encryptedPayload)) { - cb(ResultCryptoError, msg.getMessageId()); - return; - } - payload = encryptedPayload; +static SharedBuffer applyCompression(const SharedBuffer& uncompressedPayload, + CompressionType compressionType) { + return CompressionCodecProvider::getCodec(compressionType).encode(uncompressedPayload); +} - if (payloadSize > ClientConnection::getMaxMessageSize() && !chunkingEnabled_) { - LOG_DEBUG(getName() << " - compressed Message payload size" << payloadSize << "cannot exceed " - << ClientConnection::getMaxMessageSize() << " bytes"); - cb(ResultMessageTooBig, msg.getMessageId()); - return; - } +void ProducerImpl::sendAsync(const Message& msg, SendCallback callback) { + if (!isValidProducerState(callback)) { + return; } - const auto messageId = msg.getMessageId(); - if (msg.impl_->metadata.has_producer_name()) { - // Message had already been sent before - releaseSemaphore(payloadSize); - cb(ResultInvalidMessage, messageId); + const auto& uncompressedPayload = msg.impl_->payload; + uint32_t uncompressedSize = uncompressedPayload.readableBytes(); + if (!canEnqueueRequest(callback, uncompressedSize)) { return; } - const int totalChunks = - conf_.getBatchingEnabled() - ? 1 - : getNumOfChunks(uncompressedSize, static_cast(ClientConnection::getMaxMessageSize())); - // Each chunk should be sent individually, so try to acquire extra permits for chunks. - for (int i = 0; i < totalChunks - 1; i++) { - if (!canEnqueueRequest(0, messageId, cb)) { - return; - } + const bool compressed = !canAddToBatch(msg); + const auto payload = + compressed ? uncompressedPayload : applyCompression(uncompressedPayload, conf_.getCompressionType()); + const auto compressedSize = static_cast(payload.readableBytes()); + const auto maxMessageSize = static_cast(ClientConnection::getMaxMessageSize()); + if (compressedSize > ClientConnection::getMaxMessageSize() && !chunkingEnabled_) { + LOG_DEBUG(getName() << " - compressed Message payload size " << payload.readableBytes() + << " cannot exceed " << ClientConnection::getMaxMessageSize() << " bytes"); + releaseSemaphore(uncompressedSize); + callback(ResultMessageTooBig, {}); + return; } - // Reserve a spot in the messages queue before acquiring the ProducerImpl - // mutex. When the queue is full, this call will block until a spot is - // available. - if (!canEnqueueRequest(payloadSize, messageId, cb)) { + auto& msgMetadata = msg.impl_->metadata; + if (!msgMetadata.has_replicated_from() && msgMetadata.has_producer_name()) { + releaseSemaphore(uncompressedSize); + callback(ResultInvalidMessage, {}); return; } - Lock lock(mutex_); - // producers may be lazily starting and be in the pending state - if (state_ != Ready && state_ != Pending) { - lock.unlock(); - releaseSemaphore(payloadSize); - cb(ResultAlreadyClosed, messageId); - return; + const int totalChunks = + canAddToBatch(msg) ? 1 : getNumOfChunks(uncompressedSize, ClientConnection::getMaxMessageSize()); + // Each chunk should be sent individually, so try to acquire extra permits for chunks. + for (int i = 0; i < (totalChunks - 1); i++) { + if (!canEnqueueRequest(callback, 0 /* The memory has already reserved */)) { + releaseSemaphore(uncompressedSize); + return; + } } - uint64_t sequenceId; - if (!msg.impl_->metadata.has_sequence_id()) { + Lock lock(mutex_); + long sequenceId; + if (!msgMetadata.has_sequence_id()) { sequenceId = msgSequenceGenerator_++; } else { - sequenceId = msg.impl_->metadata.sequence_id(); + sequenceId = msgMetadata.sequence_id(); } setMessageMetadata(msg, sequenceId, uncompressedSize); - const std::string uuid = (totalChunks > 1) ? (producerName_ + "-" + std::to_string(sequenceId)) : ""; - const std::string schemaVersion = (totalChunks > 1 && msg.impl_->metadata.has_schema_version()) - ? msg.impl_->metadata.schema_version() - : ""; - int readStartIndex = 0; - for (int chunkId = 0; chunkId < totalChunks; chunkId++) { - // The schema version might change after serializeAndSendMessage(), so we need to reset it. - if (chunkId > 0 && msg.impl_->metadata.has_schema_version()) { - msg.impl_->metadata.set_schema_version(schemaVersion); - } - serializeAndSendMessage(msg, msg.impl_->payload, sequenceId, uuid, chunkId, totalChunks, - readStartIndex, ClientConnection::getMaxMessageSize(), payload, compressed, - payloadSize, uncompressedSize, cb); - readStartIndex = (chunkId + 1) * ClientConnection::getMaxMessageSize(); - } -} - -void ProducerImpl::serializeAndSendMessage(const Message& msg, SharedBuffer& payload, uint64_t sequenceId, - const std::string& uuid, int chunkId, int totalChunks, - int readStartIndex, int chunkMaxSizeInBytes, - SharedBuffer& compressedPayload, bool compressed, - int compressedPayloadSize, int uncompressedSize, SendCallback cb) { - // TODO: implement the real logic - if (batchMessageContainer_ && !msg.impl_->metadata.has_deliver_at_time()) { + producerStatsBasePtr_->messageSent(msg); + auto self = shared_from_this(); + const auto now = boost::posix_time::microsec_clock::universal_time(); + SendCallback cb = [this, self, now, callback](Result result, const MessageId& messageId) { + producerStatsBasePtr_->messageReceived(result, now); + callback(result, messageId); + }; + + if (canAddToBatch(msg)) { // Batching is enabled and the message is not delayed if (!batchMessageContainer_->hasEnoughSpace(msg)) { batchMessageAndSend().complete(); @@ -476,12 +457,45 @@ void ProducerImpl::serializeAndSendMessage(const Message& msg, SharedBuffer& pay if (isFull) { auto failures = batchMessageAndSend(); - // lock.unlock(); + lock.unlock(); failures.complete(); } } else { - sendMessage(OpSendMsg{msg, cb, producerId_, sequenceId, conf_.getSendTimeout(), 1, - static_cast(compressedPayloadSize)}); + const bool sendChunks = (totalChunks > 1); + if (sendChunks) { + msgMetadata.set_uuid(producerName_ + "-" + std::to_string(sequenceId)); + msgMetadata.set_num_chunks_from_msg(totalChunks); + msgMetadata.set_total_chunk_msg_size(compressedSize); + } + + int beginIndex = 0; + for (int chunkId = 0; chunkId < totalChunks; chunkId++) { + if (sendChunks) { + msgMetadata.set_chunk_id(chunkId); + } + const uint32_t endIndex = std::min(compressedSize, beginIndex + maxMessageSize); + auto chunkedPayload = payload.slice(beginIndex, endIndex - beginIndex); + beginIndex = endIndex; + + SharedBuffer encryptedPayload; + if (!encryptMessage(msgMetadata, chunkedPayload, encryptedPayload)) { + releaseSemaphore(uncompressedSize); + cb(ResultCryptoError, {}); + return; + } + + OpSendMsg op{msg, + cb, + producerId_, + static_cast(sequenceId), + conf_.getSendTimeout(), + 1, + uncompressedSize}; + if (sendChunks) { + op.setChunkId(chunkId).setTotalChunks(totalChunks); + } + sendMessage(op); + } } } @@ -518,8 +532,7 @@ Result ProducerImpl::canEnqueueRequest(uint32_t payloadSize) { } } -bool ProducerImpl::canEnqueueRequest(uint32_t payloadSize, const MessageId& messageId, - const SendCallback& cb) { +bool ProducerImpl::canEnqueueRequest(const SendCallback& callback, uint32_t payloadSize) { const auto result = canEnqueueRequest(payloadSize); if (result != ResultOk) { // If queue is full sending the batch immediately, no point waiting till batchMessagetimeout @@ -531,7 +544,7 @@ bool ProducerImpl::canEnqueueRequest(uint32_t payloadSize, const MessageId& mess failures.complete(); } - cb(result, messageId); + callback(result, {}); } return result == ResultOk; } @@ -846,7 +859,8 @@ bool ProducerImpl::ackReceived(uint64_t sequenceId, MessageId& rawMessageId) { pendingMessagesQueue_.pop_front(); lock.unlock(); - if (op.sendCallback_) { + // If message is chunked, then call callback only on last chunk + if (op.totalChunks_ <= 1 || (op.chunkId_ == op.totalChunks_ - 1)) { try { op.sendCallback_(ResultOk, messageId); } catch (const std::exception& e) { diff --git a/pulsar-client-cpp/lib/ProducerImpl.h b/pulsar-client-cpp/lib/ProducerImpl.h index 05b81e7b69e28..a69d691ce7faf 100644 --- a/pulsar-client-cpp/lib/ProducerImpl.h +++ b/pulsar-client-cpp/lib/ProducerImpl.h @@ -118,8 +118,6 @@ class ProducerImpl : public HandlerBase, void handleCreateProducer(const ClientConnectionPtr& cnx, Result result, const ResponseData& responseData); - void statsCallBackHandler(Result, const MessageId&, SendCallback, boost::posix_time::ptime); - void handleClose(Result result, ResultCallback callback, ProducerImplPtr producer); void resendMessages(ClientConnectionPtr cnx); @@ -139,21 +137,18 @@ class ProducerImpl : public HandlerBase, Result canEnqueueRequest(uint32_t payloadSize); /** - * It calls the previous overloaded method. If the result is not ResultOk, `cb` will be completed with - * `result` and `messageId` and `batchMessageAndSend` will be called to send all pending messages. + * It calls the previous overloaded method. If the result is not ResultOk, `batchMessageAndSend` will be + * called to send all pending messages. Then `callback` will be completed with `result` and a default + * MessageId. */ - bool canEnqueueRequest(uint32_t payloadSize, const MessageId& messageId, const SendCallback& cb); - - void serializeAndSendMessage(const Message& msg, SharedBuffer& payload, uint64_t sequenceId, - const std::string& uuid, int chunkId, int totalChunks, int readStartIndex, - int chunkMaxSizeInBytes, SharedBuffer& compressedPayload, bool compressed, - int compressedPayloadSize, int uncompressedSize, SendCallback cb); + bool canEnqueueRequest(const SendCallback& callback, uint32_t size); void releaseSemaphore(uint32_t payloadSize); void releaseSemaphoreForSendOp(const OpSendMsg& op); void cancelTimers(); + bool isValidProducerState(const SendCallback& callback); bool canAddToBatch(const Message& msg) const; typedef std::unique_lock Lock; @@ -163,7 +158,7 @@ class ProducerImpl : public HandlerBase, std::unique_ptr semaphore_; MessageQueue pendingMessagesQueue_; - int32_t partition_; // -1 if topic is non-partitioned + const int32_t partition_; // -1 if topic is non-partitioned std::string producerName_; bool userProvidedProducerName_; std::string producerStr_; diff --git a/pulsar-client-cpp/lib/SharedBuffer.h b/pulsar-client-cpp/lib/SharedBuffer.h index d54429984b659..e99edb17fb278 100644 --- a/pulsar-client-cpp/lib/SharedBuffer.h +++ b/pulsar-client-cpp/lib/SharedBuffer.h @@ -94,13 +94,13 @@ class SharedBuffer { /** * Return a shared buffer that include a portion of current buffer. No memory is copied */ - SharedBuffer slice(uint32_t offset) { + SharedBuffer slice(uint32_t offset) const { SharedBuffer buf(*this); buf.consume(offset); return buf; } - SharedBuffer slice(uint32_t offset, uint32_t length) { + SharedBuffer slice(uint32_t offset, uint32_t length) const { SharedBuffer buf(*this); buf.consume(offset); assert(buf.readableBytes() >= length); diff --git a/pulsar-client-cpp/lib/stats/ProducerStatsBase.h b/pulsar-client-cpp/lib/stats/ProducerStatsBase.h index 494303b5dfe82..0ae16d1769c7c 100644 --- a/pulsar-client-cpp/lib/stats/ProducerStatsBase.h +++ b/pulsar-client-cpp/lib/stats/ProducerStatsBase.h @@ -27,7 +27,7 @@ namespace pulsar { class ProducerStatsBase { public: virtual void messageSent(const Message& msg) = 0; - virtual void messageReceived(Result&, boost::posix_time::ptime&) = 0; + virtual void messageReceived(Result, const boost::posix_time::ptime&) = 0; virtual ~ProducerStatsBase(){}; }; diff --git a/pulsar-client-cpp/lib/stats/ProducerStatsDisabled.h b/pulsar-client-cpp/lib/stats/ProducerStatsDisabled.h index f81c8aa1a7944..6568c07487719 100644 --- a/pulsar-client-cpp/lib/stats/ProducerStatsDisabled.h +++ b/pulsar-client-cpp/lib/stats/ProducerStatsDisabled.h @@ -25,7 +25,7 @@ namespace pulsar { class ProducerStatsDisabled : public ProducerStatsBase { public: virtual void messageSent(const Message& msg){}; - virtual void messageReceived(Result&, boost::posix_time::ptime&){}; + virtual void messageReceived(Result, const boost::posix_time::ptime&){}; }; } // namespace pulsar #endif // PULSAR_PRODUCER_STATS_DISABLED_HEADER diff --git a/pulsar-client-cpp/lib/stats/ProducerStatsImpl.cc b/pulsar-client-cpp/lib/stats/ProducerStatsImpl.cc index af7ae4b9c0440..811c0e16753fe 100644 --- a/pulsar-client-cpp/lib/stats/ProducerStatsImpl.cc +++ b/pulsar-client-cpp/lib/stats/ProducerStatsImpl.cc @@ -93,7 +93,7 @@ void ProducerStatsImpl::messageSent(const Message& msg) { totalBytesSent_ += msg.getLength(); } -void ProducerStatsImpl::messageReceived(Result& res, boost::posix_time::ptime& publishTime) { +void ProducerStatsImpl::messageReceived(Result res, const boost::posix_time::ptime& publishTime) { boost::posix_time::ptime currentTime = boost::posix_time::microsec_clock::universal_time(); double diffInMicros = (currentTime - publishTime).total_microseconds(); Lock lock(mutex_); diff --git a/pulsar-client-cpp/lib/stats/ProducerStatsImpl.h b/pulsar-client-cpp/lib/stats/ProducerStatsImpl.h index e82628b140abe..27ffacc81a519 100644 --- a/pulsar-client-cpp/lib/stats/ProducerStatsImpl.h +++ b/pulsar-client-cpp/lib/stats/ProducerStatsImpl.h @@ -82,7 +82,7 @@ class ProducerStatsImpl : public std::enable_shared_from_this void messageSent(const Message&); - void messageReceived(Result&, boost::posix_time::ptime&); + void messageReceived(Result, const boost::posix_time::ptime&); ~ProducerStatsImpl(); From 22b3453504a3b9559ffb9a2b4311cfece94ad8cf Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Wed, 22 Dec 2021 18:07:35 +0800 Subject: [PATCH 06/27] Check whether the callback should be triggered in sendMessage --- pulsar-client-cpp/lib/OpSendMsg.h | 12 ------------ pulsar-client-cpp/lib/ProducerImpl.cc | 17 ++++------------- 2 files changed, 4 insertions(+), 25 deletions(-) diff --git a/pulsar-client-cpp/lib/OpSendMsg.h b/pulsar-client-cpp/lib/OpSendMsg.h index dac04004b8da1..70b880cf5179c 100644 --- a/pulsar-client-cpp/lib/OpSendMsg.h +++ b/pulsar-client-cpp/lib/OpSendMsg.h @@ -36,8 +36,6 @@ struct OpSendMsg { boost::posix_time::ptime timeout_; uint32_t messagesCount_; uint64_t messagesSize_; - int totalChunks_ = 0; - int chunkId_ = -1; OpSendMsg() = default; @@ -50,16 +48,6 @@ struct OpSendMsg { timeout_(TimeUtils::now() + milliseconds(sendTimeoutMs)), messagesCount_(messagesCount), messagesSize_(messagesSize) {} - - OpSendMsg& setTotalChunks(int totalChunks) { - totalChunks_ = totalChunks; - return *this; - } - - OpSendMsg& setChunkId(int chunkId) { - chunkId_ = chunkId; - return *this; - } }; } // namespace pulsar diff --git a/pulsar-client-cpp/lib/ProducerImpl.cc b/pulsar-client-cpp/lib/ProducerImpl.cc index 4b59eb5627932..2f99822525cab 100644 --- a/pulsar-client-cpp/lib/ProducerImpl.cc +++ b/pulsar-client-cpp/lib/ProducerImpl.cc @@ -484,17 +484,9 @@ void ProducerImpl::sendAsync(const Message& msg, SendCallback callback) { return; } - OpSendMsg op{msg, - cb, - producerId_, - static_cast(sequenceId), - conf_.getSendTimeout(), - 1, - uncompressedSize}; - if (sendChunks) { - op.setChunkId(chunkId).setTotalChunks(totalChunks); - } - sendMessage(op); + sendMessage(OpSendMsg{msg, (chunkId == totalChunks - 1) ? cb : nullptr, producerId_, + static_cast(sequenceId), conf_.getSendTimeout(), 1, + uncompressedSize}); } } } @@ -859,8 +851,7 @@ bool ProducerImpl::ackReceived(uint64_t sequenceId, MessageId& rawMessageId) { pendingMessagesQueue_.pop_front(); lock.unlock(); - // If message is chunked, then call callback only on last chunk - if (op.totalChunks_ <= 1 || (op.chunkId_ == op.totalChunks_ - 1)) { + if (op.sendCallback_) { try { op.sendCallback_(ResultOk, messageId); } catch (const std::exception& e) { From 644becba3bd146440b1341909b502e03c5a16ac1 Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Thu, 23 Dec 2021 10:40:26 +0800 Subject: [PATCH 07/27] Combine callback and releaseSemaphore --- .../lib/MemoryLimitController.cc | 6 +- pulsar-client-cpp/lib/ProducerImpl.cc | 62 ++++++++++--------- pulsar-client-cpp/lib/ProducerImpl.h | 7 --- 3 files changed, 37 insertions(+), 38 deletions(-) diff --git a/pulsar-client-cpp/lib/MemoryLimitController.cc b/pulsar-client-cpp/lib/MemoryLimitController.cc index 81578c9e3f4fb..4a23f8b1a1d0c 100644 --- a/pulsar-client-cpp/lib/MemoryLimitController.cc +++ b/pulsar-client-cpp/lib/MemoryLimitController.cc @@ -25,6 +25,10 @@ MemoryLimitController::MemoryLimitController(uint64_t memoryLimit) : memoryLimit_(memoryLimit), currentUsage_(0), mutex_(), condition_() {} bool MemoryLimitController::tryReserveMemory(uint64_t size) { + // Avoid CAS operation when size is 0 + if (size == 0) { + return true; + } while (true) { uint64_t current = currentUsage_; uint64_t newUsage = current + size; @@ -66,4 +70,4 @@ void MemoryLimitController::releaseMemory(uint64_t size) { uint64_t MemoryLimitController::currentUsage() const { return currentUsage_; } -} // namespace pulsar \ No newline at end of file +} // namespace pulsar diff --git a/pulsar-client-cpp/lib/ProducerImpl.cc b/pulsar-client-cpp/lib/ProducerImpl.cc index 2f99822525cab..4b3081bee6ac4 100644 --- a/pulsar-client-cpp/lib/ProducerImpl.cc +++ b/pulsar-client-cpp/lib/ProducerImpl.cc @@ -389,28 +389,45 @@ void ProducerImpl::sendAsync(const Message& msg, SendCallback callback) { } const auto& uncompressedPayload = msg.impl_->payload; - uint32_t uncompressedSize = uncompressedPayload.readableBytes(); - if (!canEnqueueRequest(callback, uncompressedSize)) { + const uint32_t uncompressedSize = uncompressedPayload.readableBytes(); + const auto result = canEnqueueRequest(uncompressedSize); + if (result != ResultOk) { + // If queue is full sending the batch immediately, no point waiting till batchMessagetimeout + if (batchMessageContainer_) { + LOG_DEBUG(getName() << " - sending batch message immediately"); + Lock lock(mutex_); + auto failures = batchMessageAndSend(); + lock.unlock(); + failures.complete(); + } + + callback(result, {}); return; } + // We have already reserved a spot, so if we need to early return for failed result, we should release the + // semaphore and memory first. + auto handleFailedResult = [this, uncompressedSize, callback](Result result) { + releaseSemaphore(uncompressedSize); // it releases the memory as well + callback(result, {}); + }; + const bool compressed = !canAddToBatch(msg); const auto payload = compressed ? uncompressedPayload : applyCompression(uncompressedPayload, conf_.getCompressionType()); const auto compressedSize = static_cast(payload.readableBytes()); const auto maxMessageSize = static_cast(ClientConnection::getMaxMessageSize()); + if (compressedSize > ClientConnection::getMaxMessageSize() && !chunkingEnabled_) { LOG_DEBUG(getName() << " - compressed Message payload size " << payload.readableBytes() << " cannot exceed " << ClientConnection::getMaxMessageSize() << " bytes"); - releaseSemaphore(uncompressedSize); - callback(ResultMessageTooBig, {}); + handleFailedResult(ResultMessageTooBig); return; } auto& msgMetadata = msg.impl_->metadata; if (!msgMetadata.has_replicated_from() && msgMetadata.has_producer_name()) { - releaseSemaphore(uncompressedSize); - callback(ResultInvalidMessage, {}); + handleFailedResult(ResultInvalidMessage); return; } @@ -418,8 +435,9 @@ void ProducerImpl::sendAsync(const Message& msg, SendCallback callback) { canAddToBatch(msg) ? 1 : getNumOfChunks(uncompressedSize, ClientConnection::getMaxMessageSize()); // Each chunk should be sent individually, so try to acquire extra permits for chunks. for (int i = 0; i < (totalChunks - 1); i++) { - if (!canEnqueueRequest(callback, 0 /* The memory has already reserved */)) { - releaseSemaphore(uncompressedSize); + const auto result = canEnqueueRequest(0); // size is 0 because the memory has already reserved + if (result != ResultOk) { + handleFailedResult(result); return; } } @@ -436,7 +454,8 @@ void ProducerImpl::sendAsync(const Message& msg, SendCallback callback) { producerStatsBasePtr_->messageSent(msg); auto self = shared_from_this(); const auto now = boost::posix_time::microsec_clock::universal_time(); - SendCallback cb = [this, self, now, callback](Result result, const MessageId& messageId) { + SendCallback callbackWithStatsUpdate = [this, self, now, callback](Result result, + const MessageId& messageId) { producerStatsBasePtr_->messageReceived(result, now); callback(result, messageId); }; @@ -447,7 +466,7 @@ void ProducerImpl::sendAsync(const Message& msg, SendCallback callback) { batchMessageAndSend().complete(); } bool isFirstMessage = batchMessageContainer_->isFirstMessageToAdd(msg); - bool isFull = batchMessageContainer_->add(msg, cb); + bool isFull = batchMessageContainer_->add(msg, callbackWithStatsUpdate); if (isFirstMessage) { batchTimer_->expires_from_now( boost::posix_time::milliseconds(conf_.getBatchingMaxPublishDelayMs())); @@ -480,12 +499,12 @@ void ProducerImpl::sendAsync(const Message& msg, SendCallback callback) { SharedBuffer encryptedPayload; if (!encryptMessage(msgMetadata, chunkedPayload, encryptedPayload)) { releaseSemaphore(uncompressedSize); - cb(ResultCryptoError, {}); + callbackWithStatsUpdate(ResultCryptoError, {}); return; } - sendMessage(OpSendMsg{msg, (chunkId == totalChunks - 1) ? cb : nullptr, producerId_, - static_cast(sequenceId), conf_.getSendTimeout(), 1, + sendMessage(OpSendMsg{msg, (chunkId == totalChunks - 1) ? callbackWithStatsUpdate : nullptr, + producerId_, static_cast(sequenceId), conf_.getSendTimeout(), 1, uncompressedSize}); } } @@ -524,23 +543,6 @@ Result ProducerImpl::canEnqueueRequest(uint32_t payloadSize) { } } -bool ProducerImpl::canEnqueueRequest(const SendCallback& callback, uint32_t payloadSize) { - const auto result = canEnqueueRequest(payloadSize); - if (result != ResultOk) { - // If queue is full sending the batch immediately, no point waiting till batchMessagetimeout - if (batchMessageContainer_) { - LOG_DEBUG(getName() << " - sending batch message immediately"); - Lock lock(mutex_); - auto failures = batchMessageAndSend(); - lock.unlock(); - failures.complete(); - } - - callback(result, {}); - } - return result == ResultOk; -} - void ProducerImpl::releaseSemaphore(uint32_t payloadSize) { if (semaphore_) { semaphore_->release(); diff --git a/pulsar-client-cpp/lib/ProducerImpl.h b/pulsar-client-cpp/lib/ProducerImpl.h index a69d691ce7faf..42d4e628d709e 100644 --- a/pulsar-client-cpp/lib/ProducerImpl.h +++ b/pulsar-client-cpp/lib/ProducerImpl.h @@ -136,13 +136,6 @@ class ProducerImpl : public HandlerBase, */ Result canEnqueueRequest(uint32_t payloadSize); - /** - * It calls the previous overloaded method. If the result is not ResultOk, `batchMessageAndSend` will be - * called to send all pending messages. Then `callback` will be completed with `result` and a default - * MessageId. - */ - bool canEnqueueRequest(const SendCallback& callback, uint32_t size); - void releaseSemaphore(uint32_t payloadSize); void releaseSemaphoreForSendOp(const OpSendMsg& op); From 4f2cd083a07a6139b40d7e84dc177cdfdbbaf1ec Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Thu, 23 Dec 2021 11:20:50 +0800 Subject: [PATCH 08/27] Wrap the send callback with stats update --- pulsar-client-cpp/lib/ProducerImpl.cc | 36 +++++++++++++-------------- pulsar-client-cpp/lib/ProducerImpl.h | 4 ++- 2 files changed, 21 insertions(+), 19 deletions(-) diff --git a/pulsar-client-cpp/lib/ProducerImpl.cc b/pulsar-client-cpp/lib/ProducerImpl.cc index 4b3081bee6ac4..6eef8670d054a 100644 --- a/pulsar-client-cpp/lib/ProducerImpl.cc +++ b/pulsar-client-cpp/lib/ProducerImpl.cc @@ -351,7 +351,7 @@ void ProducerImpl::triggerFlush() { } } -bool ProducerImpl::isValidProducerState(const SendCallback& callback) { +bool ProducerImpl::isValidProducerState(const SendCallback& callback) const { Lock lock(mutex_); const auto state = state_; lock.unlock(); @@ -384,6 +384,17 @@ static SharedBuffer applyCompression(const SharedBuffer& uncompressedPayload, } void ProducerImpl::sendAsync(const Message& msg, SendCallback callback) { + producerStatsBasePtr_->messageSent(msg); + + const auto now = boost::posix_time::microsec_clock::universal_time(); + auto self = shared_from_this(); + sendAsyncWithStatsUpdate(msg, [this, self, now, callback](Result result, const MessageId& messageId) { + producerStatsBasePtr_->messageReceived(result, now); + callback(result, messageId); + }); +} + +void ProducerImpl::sendAsyncWithStatsUpdate(const Message& msg, const SendCallback& callback) { if (!isValidProducerState(callback)) { return; } @@ -407,7 +418,7 @@ void ProducerImpl::sendAsync(const Message& msg, SendCallback callback) { // We have already reserved a spot, so if we need to early return for failed result, we should release the // semaphore and memory first. - auto handleFailedResult = [this, uncompressedSize, callback](Result result) { + const auto handleFailedResult = [this, uncompressedSize, callback](Result result) { releaseSemaphore(uncompressedSize); // it releases the memory as well callback(result, {}); }; @@ -443,7 +454,7 @@ void ProducerImpl::sendAsync(const Message& msg, SendCallback callback) { } Lock lock(mutex_); - long sequenceId; + uint64_t sequenceId; if (!msgMetadata.has_sequence_id()) { sequenceId = msgSequenceGenerator_++; } else { @@ -451,22 +462,13 @@ void ProducerImpl::sendAsync(const Message& msg, SendCallback callback) { } setMessageMetadata(msg, sequenceId, uncompressedSize); - producerStatsBasePtr_->messageSent(msg); - auto self = shared_from_this(); - const auto now = boost::posix_time::microsec_clock::universal_time(); - SendCallback callbackWithStatsUpdate = [this, self, now, callback](Result result, - const MessageId& messageId) { - producerStatsBasePtr_->messageReceived(result, now); - callback(result, messageId); - }; - if (canAddToBatch(msg)) { // Batching is enabled and the message is not delayed if (!batchMessageContainer_->hasEnoughSpace(msg)) { batchMessageAndSend().complete(); } bool isFirstMessage = batchMessageContainer_->isFirstMessageToAdd(msg); - bool isFull = batchMessageContainer_->add(msg, callbackWithStatsUpdate); + bool isFull = batchMessageContainer_->add(msg, callback); if (isFirstMessage) { batchTimer_->expires_from_now( boost::posix_time::milliseconds(conf_.getBatchingMaxPublishDelayMs())); @@ -498,14 +500,12 @@ void ProducerImpl::sendAsync(const Message& msg, SendCallback callback) { SharedBuffer encryptedPayload; if (!encryptMessage(msgMetadata, chunkedPayload, encryptedPayload)) { - releaseSemaphore(uncompressedSize); - callbackWithStatsUpdate(ResultCryptoError, {}); + handleFailedResult(ResultCryptoError); return; } - sendMessage(OpSendMsg{msg, (chunkId == totalChunks - 1) ? callbackWithStatsUpdate : nullptr, - producerId_, static_cast(sequenceId), conf_.getSendTimeout(), 1, - uncompressedSize}); + sendMessage(OpSendMsg{msg, (chunkId == totalChunks - 1) ? callback : nullptr, producerId_, + sequenceId, conf_.getSendTimeout(), 1, uncompressedSize}); } } } diff --git a/pulsar-client-cpp/lib/ProducerImpl.h b/pulsar-client-cpp/lib/ProducerImpl.h index 42d4e628d709e..3dccf911f93b9 100644 --- a/pulsar-client-cpp/lib/ProducerImpl.h +++ b/pulsar-client-cpp/lib/ProducerImpl.h @@ -126,6 +126,8 @@ class ProducerImpl : public HandlerBase, bool encryptMessage(proto::MessageMetadata& metadata, SharedBuffer& payload, SharedBuffer& encryptedPayload); + void sendAsyncWithStatsUpdate(const Message& msg, const SendCallback& callback); + /** * Reserve a spot in the messages queue before acquiring the ProducerImpl mutex. When the queue is full, * this call will block until a spot is available if blockIfQueueIsFull is true. Otherwise, it will return @@ -141,7 +143,7 @@ class ProducerImpl : public HandlerBase, void cancelTimers(); - bool isValidProducerState(const SendCallback& callback); + bool isValidProducerState(const SendCallback& callback) const; bool canAddToBatch(const Message& msg) const; typedef std::unique_lock Lock; From b97a896f2f5f16dc3845ab5251012d2ebd7a01d3 Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Thu, 23 Dec 2021 16:45:39 +0800 Subject: [PATCH 09/27] Fix incorrect isValidProducerState --- pulsar-client-cpp/lib/ProducerImpl.cc | 3 ++- pulsar-client-cpp/test-conf/standalone-ssl.conf | 5 ++++- 2 files changed, 6 insertions(+), 2 deletions(-) diff --git a/pulsar-client-cpp/lib/ProducerImpl.cc b/pulsar-client-cpp/lib/ProducerImpl.cc index 6eef8670d054a..e4f323c3adc17 100644 --- a/pulsar-client-cpp/lib/ProducerImpl.cc +++ b/pulsar-client-cpp/lib/ProducerImpl.cc @@ -360,7 +360,8 @@ bool ProducerImpl::isValidProducerState(const SendCallback& callback) const { // OK case HandlerBase::Pending: // We are OK to queue the messages on the client, it will be sent to the broker once we get the - // connectionPL + // connection + return true; case HandlerBase::Closing: case HandlerBase::Closed: callback(ResultAlreadyClosed, {}); diff --git a/pulsar-client-cpp/test-conf/standalone-ssl.conf b/pulsar-client-cpp/test-conf/standalone-ssl.conf index 90c48228f9271..8fedee7054be1 100644 --- a/pulsar-client-cpp/test-conf/standalone-ssl.conf +++ b/pulsar-client-cpp/test-conf/standalone-ssl.conf @@ -300,4 +300,7 @@ defaultNumPartitions=1 globalZookeeperServers={{ zookeeper_servers }} # Deprecated. Use brokerDeleteInactiveTopicsFrequencySeconds -brokerServicePurgeInactiveFrequencyInSeconds=60 \ No newline at end of file +brokerServicePurgeInactiveFrequencyInSeconds=60 + +# Given a specific limit of the max message size +maxMessageSize=10240 From 82679fb21fec2952933b5733381f19996029819d Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Sat, 25 Dec 2021 13:00:44 +0800 Subject: [PATCH 10/27] Fix checksum error when chunks are sent --- .../lib/BatchMessageContainerBase.cc | 3 +- pulsar-client-cpp/lib/ClientConnection.cc | 10 +++-- pulsar-client-cpp/lib/Commands.cc | 12 +++--- pulsar-client-cpp/lib/Commands.h | 3 +- pulsar-client-cpp/lib/OpSendMsg.h | 18 +++++++-- pulsar-client-cpp/lib/ProducerImpl.cc | 40 +++++++++---------- pulsar-client-cpp/lib/SharedBuffer.h | 4 +- 7 files changed, 53 insertions(+), 37 deletions(-) diff --git a/pulsar-client-cpp/lib/BatchMessageContainerBase.cc b/pulsar-client-cpp/lib/BatchMessageContainerBase.cc index 1201707e0645e..e9e6b987a107e 100644 --- a/pulsar-client-cpp/lib/BatchMessageContainerBase.cc +++ b/pulsar-client-cpp/lib/BatchMessageContainerBase.cc @@ -74,7 +74,8 @@ Result BatchMessageContainerBase::createOpSendMsgHelper(OpSendMsg& opSendMsg, return ResultMessageTooBig; } - opSendMsg.msg_.impl_ = impl; + opSendMsg.metadata_ = impl->metadata; + opSendMsg.payload_ = impl->payload; opSendMsg.sequenceId_ = impl->metadata.sequence_id(); opSendMsg.producerId_ = producerId_; opSendMsg.timeout_ = TimeUtils::now() + milliseconds(producerConfig_.getSendTimeout()); diff --git a/pulsar-client-cpp/lib/ClientConnection.cc b/pulsar-client-cpp/lib/ClientConnection.cc index 3ad6f4062f4b9..79bc1d74aa967 100644 --- a/pulsar-client-cpp/lib/ClientConnection.cc +++ b/pulsar-client-cpp/lib/ClientConnection.cc @@ -1366,8 +1366,9 @@ void ClientConnection::sendMessage(const OpSendMsg& opSend) { } void ClientConnection::sendMessageInternal(const OpSendMsg& opSend) { - PairSharedBuffer buffer = Commands::newSend(outgoingBuffer_, outgoingCmd_, opSend.producerId_, - opSend.sequenceId_, getChecksumType(), opSend.msg_); + PairSharedBuffer buffer = + Commands::newSend(outgoingBuffer_, outgoingCmd_, opSend.producerId_, opSend.sequenceId_, + getChecksumType(), opSend.metadata_, opSend.payload_); asyncWrite(buffer, customAllocWriteHandler(std::bind(&ClientConnection::handleSendPair, shared_from_this(), std::placeholders::_1))); @@ -1408,8 +1409,9 @@ void ClientConnection::sendPendingCommands() { assert(any.type() == typeid(OpSendMsg)); const OpSendMsg& op = boost::any_cast(any); - PairSharedBuffer buffer = Commands::newSend(outgoingBuffer_, outgoingCmd_, op.producerId_, - op.sequenceId_, getChecksumType(), op.msg_); + PairSharedBuffer buffer = + Commands::newSend(outgoingBuffer_, outgoingCmd_, op.producerId_, op.sequenceId_, + getChecksumType(), op.metadata_, op.payload_); asyncWrite(buffer, customAllocWriteHandler(std::bind(&ClientConnection::handleSendPair, shared_from_this(), std::placeholders::_1))); diff --git a/pulsar-client-cpp/lib/Commands.cc b/pulsar-client-cpp/lib/Commands.cc index 1094efb8eee35..472c15bed8a75 100644 --- a/pulsar-client-cpp/lib/Commands.cc +++ b/pulsar-client-cpp/lib/Commands.cc @@ -141,10 +141,8 @@ SharedBuffer Commands::newConsumerStats(uint64_t consumerId, uint64_t requestId) } PairSharedBuffer Commands::newSend(SharedBuffer& headers, BaseCommand& cmd, uint64_t producerId, - uint64_t sequenceId, ChecksumType checksumType, const Message& msg) { - const proto::MessageMetadata& metadata = msg.impl_->metadata; - SharedBuffer& payload = msg.impl_->payload; - + uint64_t sequenceId, ChecksumType checksumType, + const proto::MessageMetadata& metadata, const SharedBuffer& payload) { cmd.set_type(BaseCommand::SEND); CommandSend* send = cmd.mutable_send(); send->set_producer_id(producerId); @@ -152,6 +150,9 @@ PairSharedBuffer Commands::newSend(SharedBuffer& headers, BaseCommand& cmd, uint if (metadata.has_num_messages_in_batch()) { send->set_num_messages(metadata.num_messages_in_batch()); } + if (metadata.has_chunk_id()) { + send->set_is_chunk(true); + } // / Wire format // [TOTAL_SIZE] [CMD_SIZE][CMD] [MAGIC_NUMBER][CHECKSUM] [METADATA_SIZE][METADATA] [PAYLOAD] @@ -199,7 +200,8 @@ PairSharedBuffer Commands::newSend(SharedBuffer& headers, BaseCommand& cmd, uint int metadataStartIndex = checksumReaderIndex + checksumSize; uint32_t metadataChecksum = computeChecksum(0, headers.data() + metadataStartIndex, (writeIndex - metadataStartIndex)); - uint32_t computedChecksum = computeChecksum(metadataChecksum, payload.data(), payload.writerIndex()); + uint32_t computedChecksum = + computeChecksum(metadataChecksum, payload.data(), payload.readableBytes()); // set computed checksum headers.setWriterIndex(checksumReaderIndex); headers.writeUnsignedInt(computedChecksum); diff --git a/pulsar-client-cpp/lib/Commands.h b/pulsar-client-cpp/lib/Commands.h index e72057182ff2c..bab2211f7fa95 100644 --- a/pulsar-client-cpp/lib/Commands.h +++ b/pulsar-client-cpp/lib/Commands.h @@ -80,7 +80,8 @@ class Commands { const std::string& listenerName); static PairSharedBuffer newSend(SharedBuffer& headers, proto::BaseCommand& cmd, uint64_t producerId, - uint64_t sequenceId, ChecksumType checksumType, const Message& msg); + uint64_t sequenceId, ChecksumType checksumType, + const proto::MessageMetadata& metadata, const SharedBuffer& payload); static SharedBuffer newSubscribe(const std::string& topic, const std::string& subscription, uint64_t consumerId, uint64_t requestId, diff --git a/pulsar-client-cpp/lib/OpSendMsg.h b/pulsar-client-cpp/lib/OpSendMsg.h index 70b880cf5179c..365301be4ea95 100644 --- a/pulsar-client-cpp/lib/OpSendMsg.h +++ b/pulsar-client-cpp/lib/OpSendMsg.h @@ -29,7 +29,8 @@ namespace pulsar { struct OpSendMsg { - Message msg_; + proto::MessageMetadata metadata_; + SharedBuffer payload_; SendCallback sendCallback_; uint64_t producerId_; uint64_t sequenceId_; @@ -39,15 +40,24 @@ struct OpSendMsg { OpSendMsg() = default; - OpSendMsg(const Message& msg, const SendCallback& sendCallback, uint64_t producerId, uint64_t sequenceId, - int sendTimeoutMs, uint32_t messagesCount, uint64_t messagesSize) - : msg_(msg), + OpSendMsg(const proto::MessageMetadata& metadata, const SharedBuffer& payload, + const SendCallback& sendCallback, uint64_t producerId, uint64_t sequenceId, int sendTimeoutMs, + uint32_t messagesCount, uint64_t messagesSize) + : metadata_(metadata), // the copy happens here because OpSendMsg of chunks are constructed with the + // a shared metadata object + payload_(payload), sendCallback_(sendCallback), producerId_(producerId), sequenceId_(sequenceId), timeout_(TimeUtils::now() + milliseconds(sendTimeoutMs)), messagesCount_(messagesCount), messagesSize_(messagesSize) {} + + void complete(Result result, const MessageId& messageId) const { + if (sendCallback_) { + sendCallback_(result, messageId); + } + } }; } // namespace pulsar diff --git a/pulsar-client-cpp/lib/ProducerImpl.cc b/pulsar-client-cpp/lib/ProducerImpl.cc index e4f323c3adc17..d320a2f0be922 100644 --- a/pulsar-client-cpp/lib/ProducerImpl.cc +++ b/pulsar-client-cpp/lib/ProducerImpl.cc @@ -37,7 +37,7 @@ struct ProducerImpl::PendingCallbacks { void complete(Result result) { for (const auto& opSendMsg : opSendMsgs) { - opSendMsg.sendCallback_(result, opSendMsg.msg_.getMessageId()); + opSendMsg.complete(result, {}); } } }; @@ -431,8 +431,9 @@ void ProducerImpl::sendAsyncWithStatsUpdate(const Message& msg, const SendCallba const auto maxMessageSize = static_cast(ClientConnection::getMaxMessageSize()); if (compressedSize > ClientConnection::getMaxMessageSize() && !chunkingEnabled_) { - LOG_DEBUG(getName() << " - compressed Message payload size " << payload.readableBytes() - << " cannot exceed " << ClientConnection::getMaxMessageSize() << " bytes"); + LOG_WARN(getName() << " - compressed Message payload size " << payload.readableBytes() + << " cannot exceed " << ClientConnection::getMaxMessageSize() + << " bytes unless chunking is enabled"); handleFailedResult(ResultMessageTooBig); return; } @@ -505,8 +506,9 @@ void ProducerImpl::sendAsyncWithStatsUpdate(const Message& msg, const SendCallba return; } - sendMessage(OpSendMsg{msg, (chunkId == totalChunks - 1) ? callback : nullptr, producerId_, - sequenceId, conf_.getSendTimeout(), 1, uncompressedSize}); + sendMessage(OpSendMsg{msgMetadata, encryptedPayload, + (chunkId == totalChunks - 1) ? callback : nullptr, producerId_, sequenceId, + conf_.getSendTimeout(), 1, uncompressedSize}); } } } @@ -582,7 +584,7 @@ PendingFailures ProducerImpl::batchMessageAndSend(const FlushCallback& flushCall // we need to release the spot manually LOG_ERROR("batchMessageAndSend | Failed to createOpSendMsg: " << result); releaseSemaphoreForSendOp(opSendMsg); - failures.add(std::bind(opSendMsg.sendCallback_, result, MessageId{})); + failures.add([opSendMsg, result] { opSendMsg.complete(result, {}); }); } } else if (numBatches > 1) { std::vector opSendMsgs; @@ -596,7 +598,9 @@ PendingFailures ProducerImpl::batchMessageAndSend(const FlushCallback& flushCall LOG_ERROR("batchMessageAndSend | Failed to createOpSendMsgs[" << i << "]: " << results[i]); releaseSemaphoreForSendOp(opSendMsgs[i]); - failures.add(std::bind(opSendMsgs[i].sendCallback_, results[i], MessageId{})); + const auto& opSendMsg = opSendMsgs[i]; + const auto result = results[i]; + failures.add([opSendMsg, result] { opSendMsg.complete(result, {}); }); } } } // else numBatches is 0, do nothing @@ -610,7 +614,7 @@ PendingFailures ProducerImpl::batchMessageAndSend(const FlushCallback& flushCall // a. we have a reserved spot on the queue // b. call this function after acquiring the ProducerImpl mutex_ void ProducerImpl::sendMessage(const OpSendMsg& op) { - const auto sequenceId = op.msg_.impl_->metadata.sequence_id(); + const auto sequenceId = op.metadata_.sequence_id(); LOG_DEBUG("Inserting data to pendingMessagesQueue_"); pendingMessagesQueue_.push_back(op); @@ -807,13 +811,11 @@ bool ProducerImpl::removeCorruptMessage(uint64_t sequenceId) { LOG_DEBUG(getName() << "Remove corrupt message from queue " << sequenceId); pendingMessagesQueue_.pop_front(); lock.unlock(); - if (op.sendCallback_) { + try { // to protect from client callback exception - try { - op.sendCallback_(ResultChecksumError, op.msg_.getMessageId()); - } catch (const std::exception& e) { - LOG_ERROR(getName() << "Exception thrown from callback " << e.what()); - } + op.complete(ResultChecksumError, {}); + } catch (const std::exception& e) { + LOG_ERROR(getName() << "Exception thrown from callback " << e.what()); } releaseSemaphoreForSendOp(op); return true; @@ -854,12 +856,10 @@ bool ProducerImpl::ackReceived(uint64_t sequenceId, MessageId& rawMessageId) { pendingMessagesQueue_.pop_front(); lock.unlock(); - if (op.sendCallback_) { - try { - op.sendCallback_(ResultOk, messageId); - } catch (const std::exception& e) { - LOG_ERROR(getName() << "Exception thrown from callback " << e.what()); - } + try { + op.complete(ResultOk, messageId); + } catch (const std::exception& e) { + LOG_ERROR(getName() << "Exception thrown from callback " << e.what()); } return true; } diff --git a/pulsar-client-cpp/lib/SharedBuffer.h b/pulsar-client-cpp/lib/SharedBuffer.h index e99edb17fb278..be889a7ee97ad 100644 --- a/pulsar-client-cpp/lib/SharedBuffer.h +++ b/pulsar-client-cpp/lib/SharedBuffer.h @@ -165,7 +165,7 @@ class SharedBuffer { } // Return current writer index - uint32_t writerIndex() { return writeIdx_; } + uint32_t writerIndex() const noexcept { return writeIdx_; } // skip writerIndex void skipBytes(uint32_t size) { @@ -180,7 +180,7 @@ class SharedBuffer { } // Return current reader index - uint32_t readerIndex() { return readIdx_; } + uint32_t readerIndex() const noexcept { return readIdx_; } // set readerIndex void setReaderIndex(uint32_t index) { From 8386ee3f712f23b3ca6c569d4f5c19cfab5deb42 Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Thu, 30 Dec 2021 16:55:11 +0800 Subject: [PATCH 11/27] Add chunked configs from consumer --- .../include/pulsar/ConsumerConfiguration.h | 44 +++++++++++++++++++ .../lib/ConsumerConfiguration.cc | 17 +++++++ .../lib/ConsumerConfigurationImpl.h | 2 + .../tests/ConsumerConfigurationTest.cc | 8 ++++ 4 files changed, 71 insertions(+) diff --git a/pulsar-client-cpp/include/pulsar/ConsumerConfiguration.h b/pulsar-client-cpp/include/pulsar/ConsumerConfiguration.h index 201eba38a80d5..8ad81743bad03 100644 --- a/pulsar-client-cpp/include/pulsar/ConsumerConfiguration.h +++ b/pulsar-client-cpp/include/pulsar/ConsumerConfiguration.h @@ -437,6 +437,50 @@ class PULSAR_PUBLIC ConsumerConfiguration { */ int getPriorityLevel() const; + /** + * Consumer buffers chunk messages into memory until it receives all the chunks of the original message. + * While consuming chunk-messages, chunks from same message might not be contiguous in the stream and they + * might be mixed with other messages' chunks. so, consumer has to maintain multiple buffers to manage + * chunks coming from different messages. This mainly happens when multiple publishers are publishing + * messages on the topic concurrently or publisher failed to publish all chunks of the messages. + * + * eg: M1-C1, M2-C1, M1-C2, M2-C2 + * Here, Messages M1-C1 and M1-C2 belong to original message M1, M2-C1 and M2-C2 belong to M2 message. + * + * Buffering large number of outstanding uncompleted chunked messages can create memory pressure and it + * can be guarded by providing this maxPendingChunkedMessage threshold. Once, consumer reaches this + * threshold, it drops the outstanding unchunked-messages by silently acking or asking broker to redeliver + * later by marking it unacked. See setAutoOldestChunkedMessageOnQueueFull. + * + * Default: 100 + * + * @param maxPendingChunkedMessage the number of max pending chunked messages + */ + ConsumerConfiguration& setMaxPendingChunkedMessage(size_t maxPendingChunkedMessage); + + /** + * The associated getter of setMaxPendingChunkedMessage + */ + size_t getMaxPendingChunkedMessage() const; + + /** + * Buffering large number of outstanding uncompleted chunked messages can create memory pressure and it + * can be guarded by providing the maxPendingChunkedMessage threshold. See setMaxPendingChunkedMessage. + * Once, consumer reaches this threshold, it drops the outstanding unchunked-messages by silently acking + * if autoAckOldestChunkedMessageOnQueueFull is true else it marks them for redelivery. + * + * Default: false + * + * @param autoAckOldestChunkedMessageOnQueueFull whether to ack the discarded chunked message + */ + ConsumerConfiguration& setAutoOldestChunkedMessageOnQueueFull( + bool autoAckOldestChunkedMessageOnQueueFull); + + /** + * The associated getter of setAutoOldestChunkedMessageOnQueueFull + */ + bool isAutoOldestChunkedMessageOnQueueFull() const; + friend class PulsarWrapper; private: diff --git a/pulsar-client-cpp/lib/ConsumerConfiguration.cc b/pulsar-client-cpp/lib/ConsumerConfiguration.cc index b01e4e595b6d7..d13cb0e0c86b2 100644 --- a/pulsar-client-cpp/lib/ConsumerConfiguration.cc +++ b/pulsar-client-cpp/lib/ConsumerConfiguration.cc @@ -231,4 +231,21 @@ ConsumerConfiguration& ConsumerConfiguration::setKeySharedPolicy(KeySharedPolicy KeySharedPolicy ConsumerConfiguration::getKeySharedPolicy() const { return impl_->keySharedPolicy; } +ConsumerConfiguration& ConsumerConfiguration::setMaxPendingChunkedMessage(size_t maxPendingChunkedMessage) { + impl_->maxPendingChunkedMessage = maxPendingChunkedMessage; + return *this; +} + +size_t ConsumerConfiguration::getMaxPendingChunkedMessage() const { return impl_->maxPendingChunkedMessage; } + +ConsumerConfiguration& ConsumerConfiguration::setAutoOldestChunkedMessageOnQueueFull( + bool autoAckOldestChunkedMessageOnQueueFull) { + impl_->autoAckOldestChunkedMessageOnQueueFull = autoAckOldestChunkedMessageOnQueueFull; + return *this; +} + +bool ConsumerConfiguration::isAutoOldestChunkedMessageOnQueueFull() const { + return impl_->autoAckOldestChunkedMessageOnQueueFull; +} + } // namespace pulsar diff --git a/pulsar-client-cpp/lib/ConsumerConfigurationImpl.h b/pulsar-client-cpp/lib/ConsumerConfigurationImpl.h index 75f65a7eb8eab..9c2a4615fef43 100644 --- a/pulsar-client-cpp/lib/ConsumerConfigurationImpl.h +++ b/pulsar-client-cpp/lib/ConsumerConfigurationImpl.h @@ -50,6 +50,8 @@ struct ConsumerConfigurationImpl { std::map properties; int priorityLevel{0}; KeySharedPolicy keySharedPolicy; + size_t maxPendingChunkedMessage{100}; + bool autoAckOldestChunkedMessageOnQueueFull{false}; }; } // namespace pulsar #endif /* LIB_CONSUMERCONFIGURATIONIMPL_H_ */ diff --git a/pulsar-client-cpp/tests/ConsumerConfigurationTest.cc b/pulsar-client-cpp/tests/ConsumerConfigurationTest.cc index 199b50c0f7174..57ed0ec7c79c6 100644 --- a/pulsar-client-cpp/tests/ConsumerConfigurationTest.cc +++ b/pulsar-client-cpp/tests/ConsumerConfigurationTest.cc @@ -59,6 +59,8 @@ TEST(ConsumerConfigurationTest, testDefaultConfig) { ASSERT_EQ(conf.isReplicateSubscriptionStateEnabled(), false); ASSERT_EQ(conf.getProperties().empty(), true); ASSERT_EQ(conf.getPriorityLevel(), 0); + ASSERT_EQ(conf.getMaxPendingChunkedMessage(), 100); + ASSERT_EQ(conf.isAutoOldestChunkedMessageOnQueueFull(), false); } TEST(ConsumerConfigurationTest, testCustomConfig) { @@ -139,6 +141,12 @@ TEST(ConsumerConfigurationTest, testCustomConfig) { conf.setPriorityLevel(1); ASSERT_EQ(conf.getPriorityLevel(), 1); + + conf.setMaxPendingChunkedMessage(500); + ASSERT_EQ(conf.getMaxPendingChunkedMessage(), 500); + + conf.setAutoOldestChunkedMessageOnQueueFull(true); + ASSERT_TRUE(conf.isAutoOldestChunkedMessageOnQueueFull()); } TEST(ConsumerConfigurationTest, testReadCompactPersistentExclusive) { From 74b6b6d9d0fe9e100926734d8df94f3f30edc8f5 Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Thu, 30 Dec 2021 22:58:56 +0800 Subject: [PATCH 12/27] Support consuming chunks --- .../include/pulsar/ConsumerConfiguration.h | 2 + pulsar-client-cpp/lib/ConsumerImpl.cc | 144 +++++++++++++++--- pulsar-client-cpp/lib/ConsumerImpl.h | 74 ++++++++- 3 files changed, 201 insertions(+), 19 deletions(-) diff --git a/pulsar-client-cpp/include/pulsar/ConsumerConfiguration.h b/pulsar-client-cpp/include/pulsar/ConsumerConfiguration.h index 8ad81743bad03..2cdbf4707061e 100644 --- a/pulsar-client-cpp/include/pulsar/ConsumerConfiguration.h +++ b/pulsar-client-cpp/include/pulsar/ConsumerConfiguration.h @@ -452,6 +452,8 @@ class PULSAR_PUBLIC ConsumerConfiguration { * threshold, it drops the outstanding unchunked-messages by silently acking or asking broker to redeliver * later by marking it unacked. See setAutoOldestChunkedMessageOnQueueFull. * + * If it's zero, the pending chunked messages will not be limited. + * * Default: 100 * * @param maxPendingChunkedMessage the number of max pending chunked messages diff --git a/pulsar-client-cpp/lib/ConsumerImpl.cc b/pulsar-client-cpp/lib/ConsumerImpl.cc index 77c0fa9d52a94..6d2b4c1229fea 100644 --- a/pulsar-client-cpp/lib/ConsumerImpl.cc +++ b/pulsar-client-cpp/lib/ConsumerImpl.cc @@ -63,7 +63,9 @@ ConsumerImpl::ConsumerImpl(const ClientImplPtr client, const std::string& topic, negativeAcksTracker_(client, *this, conf), ackGroupingTrackerPtr_(std::make_shared()), readCompacted_(conf.isReadCompacted()), - lastMessageInBroker_(Optional::of(MessageId())) { + lastMessageInBroker_(Optional::of(MessageId())), + maxPendingChunkedMessage_(conf.getMaxPendingChunkedMessage()), + autoAckOldestChunkedMessageOnQueueFull_(conf.isAutoOldestChunkedMessageOnQueueFull()) { std::stringstream consumerStrStream; consumerStrStream << "[" << topic_ << ", " << subscription_ << ", " << consumerId_ << "] "; consumerStr_ = consumerStrStream.str(); @@ -308,6 +310,95 @@ void ConsumerImpl::handleUnsubscribe(Result result, ResultCallback callback) { callback(result); } +bool ConsumerImpl::processMessageChunk(SharedBuffer& payload, const proto::MessageMetadata& metadata, + const MessageId& messageId, const proto::MessageIdData& messageIdData, + const ClientConnectionPtr& cnx) { + const auto chunkId = metadata.chunk_id(); + const auto uuid = metadata.uuid(); + + Lock lock(chunkProcessMutex_); + auto it = chunkedMessagesMap_.find(uuid); + + if (chunkId == 0) { + if (it == chunkedMessagesMap_.end()) { + it = chunkedMessagesMap_ + .emplace(uuid, ChunkedMessageCtx{metadata.num_chunks_from_msg(), + metadata.total_chunk_msg_size()}) + .first; + } + pendingChunkedMessage_++; + if (maxPendingChunkedMessage_ > 0 && pendingChunkedMessage_ > maxPendingChunkedMessage_) { + removeOldestPendingChunkedMessage(); + } + pendingChunkedMessageUuidQueue_.emplace_back(uuid); + } + + auto& chunkedMsgCtx = it->second; + if (it == chunkedMessagesMap_.end() || !chunkedMsgCtx.validateChunkId(chunkId)) { + if (it == chunkedMessagesMap_.end()) { + LOG_ERROR("Received unexpected chunk, messageId: " << messageId << ", chunkId: " << chunkId); + } else { + LOG_ERROR("Received unexpected chunk, messageId: " << messageId << ", chunkId: " << chunkId + << ", ChunkedMessageCtx: " << chunkedMsgCtx); + } + chunkedMessagesMap_.erase(uuid); + lock.unlock(); + increaseAvailablePermits(cnx); + trackMessage(messageId); + return false; + } + + chunkedMsgCtx.appendChunk(chunkId, messageId, payload); + if (!chunkedMsgCtx.isCompleted()) { + lock.unlock(); + increaseAvailablePermits(cnx); + return false; + } + + LOG_DEBUG("Chunked message completed chunkId: " << chunkId << ", ChunkedMessageCtx: " << chunkedMsgCtx + << ", sequenceId: " << metadata.sequence_id()); + + removeChunkMessage(uuid, false); + return uncompressMessageIfNeeded(cnx, messageIdData, metadata, payload); +} + +// It must be called when `chunkProcessMutex_` is acquired +void ConsumerImpl::removeOldestPendingChunkedMessage() { + const int numChunksToRemove = pendingChunkedMessage_ - maxPendingChunkedMessage_; + auto it = pendingChunkedMessageUuidQueue_.begin(); + for (int i = 0; i < numChunksToRemove; i++) { + it = pendingChunkedMessageUuidQueue_.erase(it); + removeChunkMessage(*it, true); + } +} + +// It must be called when `chunkProcessMutex_` is acquired +void ConsumerImpl::removeChunkMessage(const std::string& uuid, bool processMessageId) { + pendingChunkedMessage_--; + auto it = chunkedMessagesMap_.find(uuid); + if (it == chunkedMessagesMap_.end()) { + return; + } + it = chunkedMessagesMap_.erase(it); + + if (!processMessageId) { + return; + } + const auto& chunkedMsgCtx = it->second; + for (const auto& messageId : chunkedMsgCtx.getChunkedMessageIds()) { + if (autoAckOldestChunkedMessageOnQueueFull_) { + doAcknowledgeIndividual(messageId, [uuid, messageId](Result result) { + if (result != ResultOk) { + LOG_WARN("Failed to acknowledge discarded chunk, uuid: " << uuid + << ", messageId: " << messageId); + } + }); + } else { + trackMessage(messageId); + } + } +} + void ConsumerImpl::messageReceived(const ClientConnectionPtr& cnx, const proto::CommandMessage& msg, bool& isChecksumValid, proto::MessageMetadata& metadata, SharedBuffer& payload) { @@ -318,17 +409,36 @@ void ConsumerImpl::messageReceived(const ClientConnectionPtr& cnx, const proto:: return; } - if (!uncompressMessageIfNeeded(cnx, msg, metadata, payload)) { - // Message was discarded on decompression error - return; - } - if (!isChecksumValid) { // Message discarded for checksum error discardCorruptedMessage(cnx, msg.message_id(), proto::CommandAck::ChecksumMismatch); return; } + const bool isMessageDecryptable = + metadata.encryption_keys_size() <= 0 || config_.getCryptoKeyReader().get() || + config_.getCryptoFailureAction() == ConsumerCryptoFailureAction::CONSUME; + + const bool isChunkedMessage = metadata.num_chunks_from_msg() > 1 && + config_.getConsumerType() != ConsumerType::ConsumerShared && + config_.getConsumerType() != ConsumerType::ConsumerKeyShared; + if (isMessageDecryptable && !isChunkedMessage) { + if (!uncompressMessageIfNeeded(cnx, msg.message_id(), metadata, payload)) { + // Message was discarded on decompression error + return; + } + } + + // Only a non-batched messages can be a chunk + if (!metadata.has_num_messages_in_batch() && isChunkedMessage) { + const auto& messageIdData = msg.message_id(); + MessageId messageId(messageIdData.partition(), messageIdData.ledgerid(), messageIdData.entryid(), + messageIdData.batch_index()); + if (!processMessageChunk(payload, metadata, messageId, messageIdData, cnx)) { + return; + } + } + Message m(msg, metadata, payload, partitionIndex_); m.impl_->cnx_ = cnx.get(); m.impl_->setTopicName(topic_); @@ -528,7 +638,8 @@ bool ConsumerImpl::decryptMessageIfNeeded(const ClientConnectionPtr& cnx, const return false; } -bool ConsumerImpl::uncompressMessageIfNeeded(const ClientConnectionPtr& cnx, const proto::CommandMessage& msg, +bool ConsumerImpl::uncompressMessageIfNeeded(const ClientConnectionPtr& cnx, + const proto::MessageIdData& messageIdData, const proto::MessageMetadata& metadata, SharedBuffer& payload) { if (!metadata.has_compression()) { return true; @@ -542,9 +653,8 @@ bool ConsumerImpl::uncompressMessageIfNeeded(const ClientConnectionPtr& cnx, con if (payloadSize > ClientConnection::getMaxMessageSize()) { // Uncompressed size is itself corrupted since it cannot be bigger than the MaxMessageSize LOG_ERROR(getName() << "Got corrupted payload message size " << payloadSize // - << " at " << msg.message_id().ledgerid() << ":" - << msg.message_id().entryid()); - discardCorruptedMessage(cnx, msg.message_id(), proto::CommandAck::UncompressedSizeCorruption); + << " at " << messageIdData.ledgerid() << ":" << messageIdData.entryid()); + discardCorruptedMessage(cnx, messageIdData, proto::CommandAck::UncompressedSizeCorruption); return false; } } else { @@ -554,8 +664,8 @@ bool ConsumerImpl::uncompressMessageIfNeeded(const ClientConnectionPtr& cnx, con if (!CompressionCodecProvider::getCodec(compressionType).decode(payload, uncompressedSize, payload)) { LOG_ERROR(getName() << "Failed to decompress message with " << uncompressedSize // - << " at " << msg.message_id().ledgerid() << ":" << msg.message_id().entryid()); - discardCorruptedMessage(cnx, msg.message_id(), proto::CommandAck::DecompressionError); + << " at " << messageIdData.ledgerid() << ":" << messageIdData.entryid()); + discardCorruptedMessage(cnx, messageIdData, proto::CommandAck::DecompressionError); return false; } @@ -584,7 +694,7 @@ void ConsumerImpl::internalListener() { // This will only happen when the connection got reset and we cleared the queue return; } - trackMessage(msg); + trackMessage(msg.getMessageId()); try { consumerStatsBasePtr_->receivedMessage(msg, ResultOk); lastDequedMessage_ = Optional::of(msg.getMessageId()); @@ -732,7 +842,7 @@ void ConsumerImpl::messageProcessed(Message& msg, bool track) { increaseAvailablePermits(currentCnx); if (track) { - trackMessage(msg); + trackMessage(msg.getMessageId()); } } @@ -1240,11 +1350,11 @@ void ConsumerImpl::setNegativeAcknowledgeEnabledForTesting(bool enabled) { negativeAcksTracker_.setEnabledForTesting(enabled); } -void ConsumerImpl::trackMessage(const Message& msg) { +void ConsumerImpl::trackMessage(const MessageId& messageId) { if (hasParent_) { - unAckedMessageTrackerPtr_->remove(msg.getMessageId()); + unAckedMessageTrackerPtr_->remove(messageId); } else { - unAckedMessageTrackerPtr_->add(msg.getMessageId()); + unAckedMessageTrackerPtr_->add(messageId); } } diff --git a/pulsar-client-cpp/lib/ConsumerImpl.h b/pulsar-client-cpp/lib/ConsumerImpl.h index 0754a89fdf5fc..98736a234fab3 100644 --- a/pulsar-client-cpp/lib/ConsumerImpl.h +++ b/pulsar-client-cpp/lib/ConsumerImpl.h @@ -157,7 +157,7 @@ class ConsumerImpl : public ConsumerImplBase, private: bool waitingForZeroQueueSizeMessage; - bool uncompressMessageIfNeeded(const ClientConnectionPtr& cnx, const proto::CommandMessage& msg, + bool uncompressMessageIfNeeded(const ClientConnectionPtr& cnx, const proto::MessageIdData& messageIdData, const proto::MessageMetadata& metadata, SharedBuffer& payload); void discardCorruptedMessage(const ClientConnectionPtr& cnx, const proto::MessageIdData& messageId, proto::CommandAck::ValidationError validationError); @@ -177,7 +177,7 @@ class ConsumerImpl : public ConsumerImplBase, void notifyPendingReceivedCallback(Result result, Message& message, const ReceiveCallback& callback); void failPendingReceiveCallback(); void setNegativeAcknowledgeEnabledForTesting(bool enabled) override; - void trackMessage(const Message& msg); + void trackMessage(const MessageId& messageId); Optional clearReceiveQueue(); @@ -227,6 +227,76 @@ class ConsumerImpl : public ConsumerImplBase, return lastMessageInBroker_.is_present() ? lastMessageInBroker_.value() : MessageId::earliest(); } + class ChunkedMessageCtx { + public: + ChunkedMessageCtx() : totalChunks_(0) {} + ChunkedMessageCtx(int totalChunks, int totalChunkMessageSize) + : totalChunks_(totalChunks), chunkedMsgBuffer_(SharedBuffer::allocate(totalChunkMessageSize)) { + chunkedMessageIds_.reserve(totalChunks); + } + + ChunkedMessageCtx(const ChunkedMessageCtx&) = delete; + ChunkedMessageCtx(ChunkedMessageCtx&& rhs) noexcept = default; + + bool validateChunkId(int chunkId) const noexcept { + return (chunkId == lastChunkedId_ + 1) && chunkId >= 0 && chunkId < totalChunks_; + } + + void appendChunk(int chunkId, const MessageId& messageId, const SharedBuffer& payload) { + lastChunkedId_ = chunkId; + chunkedMessageIds_.emplace_back(messageId); + chunkedMsgBuffer_.write(payload.data(), payload.readableBytes()); + } + + bool isCompleted() const noexcept { return lastChunkedId_ + 1 == totalChunks_; } + + const SharedBuffer& getBuffer() const noexcept { return chunkedMsgBuffer_; } + + const std::vector& getChunkedMessageIds() const noexcept { return chunkedMessageIds_; } + + friend std::ostream& operator<<(std::ostream& os, const ChunkedMessageCtx& ctx) { + return os << "total chunks: " << ctx.totalChunks_ << ", last chunk id: " << ctx.lastChunkedId_; + } + + private: + const int totalChunks_; + SharedBuffer chunkedMsgBuffer_; + std::vector chunkedMessageIds_; + int lastChunkedId_{-1}; + }; + + mutable std::mutex chunkProcessMutex_; + std::unordered_map chunkedMessagesMap_; + const size_t maxPendingChunkedMessage_; + size_t pendingChunkedMessage_{0}; + // use list here for removing uuid of expired chunks quickly + std::list pendingChunkedMessageUuidQueue_; + + // if queue size is reasonable (most of the time equal to number of producers try to publish messages + // concurrently on the topic) then it guards against broken chunked message which was not fully published + const bool autoAckOldestChunkedMessageOnQueueFull_; + + /** + * Process a chunk. If the chunk is the last chunk of a message, concatenate all buffered chunks into the + * payload. In this case, `payload` will point to the completed payload. Otherwise, the chunk will be + * buffered. + * + * @param payload the original payload, which could be modified if this method returns true + * @param metadata the message metadata + * @param messageId + * @param messageIdData + * @param cnx + * + * @return true if chunks are concatenated into a completed message payload successfully + */ + bool processMessageChunk(SharedBuffer& payload, const proto::MessageMetadata& metadata, + const MessageId& messageId, const proto::MessageIdData& messageIdData, + const ClientConnectionPtr& cnx); + + void removeOldestPendingChunkedMessage(); + + void removeChunkMessage(const std::string& uuid, bool processMessageId); + friend class PulsarFriend; // these two declared friend to access setNegativeAcknowledgeEnabledForTesting From 0b26aa7ee937d6bca214bd1cc2da4c9a9a0a10ed Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Sat, 1 Jan 2022 00:02:28 +0800 Subject: [PATCH 13/27] Fix incorrect concanated payload size --- pulsar-client-cpp/lib/ConsumerImpl.cc | 24 +++++++++++++++++------- pulsar-client-cpp/lib/ConsumerImpl.h | 16 +++++++++------- 2 files changed, 26 insertions(+), 14 deletions(-) diff --git a/pulsar-client-cpp/lib/ConsumerImpl.cc b/pulsar-client-cpp/lib/ConsumerImpl.cc index 6d2b4c1229fea..b0a8d7ce23869 100644 --- a/pulsar-client-cpp/lib/ConsumerImpl.cc +++ b/pulsar-client-cpp/lib/ConsumerImpl.cc @@ -310,9 +310,11 @@ void ConsumerImpl::handleUnsubscribe(Result result, ResultCallback callback) { callback(result); } -bool ConsumerImpl::processMessageChunk(SharedBuffer& payload, const proto::MessageMetadata& metadata, - const MessageId& messageId, const proto::MessageIdData& messageIdData, - const ClientConnectionPtr& cnx) { +Optional ConsumerImpl::processMessageChunk(const SharedBuffer& payload, + const proto::MessageMetadata& metadata, + const MessageId& messageId, + const proto::MessageIdData& messageIdData, + const ClientConnectionPtr& cnx) { const auto chunkId = metadata.chunk_id(); const auto uuid = metadata.uuid(); @@ -345,21 +347,26 @@ bool ConsumerImpl::processMessageChunk(SharedBuffer& payload, const proto::Messa lock.unlock(); increaseAvailablePermits(cnx); trackMessage(messageId); - return false; + return Optional::empty(); } chunkedMsgCtx.appendChunk(chunkId, messageId, payload); if (!chunkedMsgCtx.isCompleted()) { lock.unlock(); increaseAvailablePermits(cnx); - return false; + return Optional::empty(); } LOG_DEBUG("Chunked message completed chunkId: " << chunkId << ", ChunkedMessageCtx: " << chunkedMsgCtx << ", sequenceId: " << metadata.sequence_id()); removeChunkMessage(uuid, false); - return uncompressMessageIfNeeded(cnx, messageIdData, metadata, payload); + auto wholePayload = chunkedMsgCtx.getBuffer(); + if (uncompressMessageIfNeeded(cnx, messageIdData, metadata, wholePayload)) { + return Optional::of(wholePayload); + } else { + return Optional::empty(); + } } // It must be called when `chunkProcessMutex_` is acquired @@ -434,7 +441,10 @@ void ConsumerImpl::messageReceived(const ClientConnectionPtr& cnx, const proto:: const auto& messageIdData = msg.message_id(); MessageId messageId(messageIdData.partition(), messageIdData.ledgerid(), messageIdData.entryid(), messageIdData.batch_index()); - if (!processMessageChunk(payload, metadata, messageId, messageIdData, cnx)) { + auto optionalPayload = processMessageChunk(payload, metadata, messageId, messageIdData, cnx); + if (optionalPayload.is_present()) { + payload = optionalPayload.value(); + } else { return; } } diff --git a/pulsar-client-cpp/lib/ConsumerImpl.h b/pulsar-client-cpp/lib/ConsumerImpl.h index 98736a234fab3..e71129a370cd9 100644 --- a/pulsar-client-cpp/lib/ConsumerImpl.h +++ b/pulsar-client-cpp/lib/ConsumerImpl.h @@ -278,20 +278,22 @@ class ConsumerImpl : public ConsumerImplBase, /** * Process a chunk. If the chunk is the last chunk of a message, concatenate all buffered chunks into the - * payload. In this case, `payload` will point to the completed payload. Otherwise, the chunk will be - * buffered. + * payload and return it. * - * @param payload the original payload, which could be modified if this method returns true + * @param payload the payload of a chunk * @param metadata the message metadata * @param messageId * @param messageIdData * @param cnx * - * @return true if chunks are concatenated into a completed message payload successfully + * @return the concatenated payload if chunks are concatenated into a completed message payload + * successfully, else Optional::empty() */ - bool processMessageChunk(SharedBuffer& payload, const proto::MessageMetadata& metadata, - const MessageId& messageId, const proto::MessageIdData& messageIdData, - const ClientConnectionPtr& cnx); + Optional processMessageChunk(const SharedBuffer& payload, + const proto::MessageMetadata& metadata, + const MessageId& messageId, + const proto::MessageIdData& messageIdData, + const ClientConnectionPtr& cnx); void removeOldestPendingChunkedMessage(); From db5c2bed5a843e06506aa084c21abd2b8a71a02c Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Tue, 4 Jan 2022 16:00:21 +0800 Subject: [PATCH 14/27] Add tests for chunking messages --- .../tests/MessageChunkingTest.cc | 113 ++++++++++++++++++ 1 file changed, 113 insertions(+) create mode 100644 pulsar-client-cpp/tests/MessageChunkingTest.cc diff --git a/pulsar-client-cpp/tests/MessageChunkingTest.cc b/pulsar-client-cpp/tests/MessageChunkingTest.cc new file mode 100644 index 0000000000000..15f6bf9690b6f --- /dev/null +++ b/pulsar-client-cpp/tests/MessageChunkingTest.cc @@ -0,0 +1,113 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +#include +#include + +#include +#include +#include "lib/LogUtils.h" + +DECLARE_LOG_OBJECT() + +using namespace pulsar; + +static const std::string lookupUrl = "pulsar://localhost:6650"; + +// See the `maxMessageSize` config in test-conf/standalone-ssl.conf +static constexpr size_t maxMessageSize = 10240; + +static std::string toString(CompressionType compressionType) { + switch (compressionType) { + case CompressionType::CompressionNone: + return "None"; + case CompressionType::CompressionLZ4: + return "LZ4"; + case CompressionType::CompressionZLib: + return "ZLib"; + case CompressionType::CompressionZSTD: + return "ZSTD"; + case CompressionType::CompressionSNAPPY: + return "SNAPPY"; + default: + return "Unknown (" + std::to_string(compressionType) + ")"; + } +} + +class MessageChunkingTest : public ::testing::TestWithParam { + public: + void TearDown() override { client_.close(); } + + void createProducer(const std::string& topic, Producer& producer) { + ProducerConfiguration conf; + conf.setBatchingEnabled(false); + conf.setChunkingEnabled(true); + conf.setCompressionType(GetParam()); + LOG_INFO("Create producer to topic: " << topic + << ", compression: " << toString(conf.getCompressionType())); + ASSERT_EQ(ResultOk, client_.createProducer(topic, conf, producer)); + } + + void createConsumer(const std::string& topic, Consumer& consumer) { + ASSERT_EQ(ResultOk, client_.subscribe(topic, "my-sub", consumer)); + } + + private: + Client client_{lookupUrl}; +}; + +TEST_F(MessageChunkingTest, testInvalidConfig) { + Client client(lookupUrl); + ProducerConfiguration conf; + conf.setBatchingEnabled(true); + conf.setChunkingEnabled(true); + Producer producer; + ASSERT_THROW(client.createProducer("xxx", conf, producer), std::invalid_argument); + client.close(); +} + +TEST_P(MessageChunkingTest, testEndToEnd) { + const std::string topic = + "MessageChunkingTest-EndToEnd-" + toString(GetParam()) + std::to_string(time(nullptr)); + Consumer consumer; + createConsumer(topic, consumer); + Producer producer; + createProducer(topic, producer); + + std::string largeMessage(maxMessageSize * 3, 'a'); + std::default_random_engine e(time(nullptr)); + std::uniform_int_distribution u(0, 25); + for (auto& ch : largeMessage) { + ch = 'a' + u(e); + } + + MessageId sendMessageId; + ASSERT_EQ(ResultOk, producer.send(MessageBuilder().setContent(largeMessage).build(), sendMessageId)); + LOG_INFO("Send to " << sendMessageId); + + Message msg; + ASSERT_EQ(ResultOk, consumer.receive(msg, 3000)); + LOG_INFO("Receive " << msg.getLength() << " bytes from " << msg.getMessageId()); + ASSERT_EQ(msg.getDataAsString(), largeMessage); + ASSERT_EQ(msg.getMessageId(), sendMessageId); +} + +INSTANTIATE_TEST_SUITE_P(Pulsar, MessageChunkingTest, ::testing::Values(CompressionNone), + [](const ::testing::TestParamInfo& info) { + return toString(info.param); + }); From 715c1f7330bdd443f96c9f589b2efb7706939f84 Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Tue, 4 Jan 2022 23:41:56 +0800 Subject: [PATCH 15/27] Fixed tests failure when compression is enabled --- pulsar-client-cpp/lib/ConsumerImpl.cc | 9 +++++---- pulsar-client-cpp/lib/ConsumerImpl.h | 3 ++- pulsar-client-cpp/lib/ProducerImpl.cc | 4 ++-- pulsar-client-cpp/tests/MessageChunkingTest.cc | 4 +++- 4 files changed, 12 insertions(+), 8 deletions(-) diff --git a/pulsar-client-cpp/lib/ConsumerImpl.cc b/pulsar-client-cpp/lib/ConsumerImpl.cc index b0a8d7ce23869..bebd92cc8e54e 100644 --- a/pulsar-client-cpp/lib/ConsumerImpl.cc +++ b/pulsar-client-cpp/lib/ConsumerImpl.cc @@ -362,7 +362,7 @@ Optional ConsumerImpl::processMessageChunk(const SharedBuffer& pay removeChunkMessage(uuid, false); auto wholePayload = chunkedMsgCtx.getBuffer(); - if (uncompressMessageIfNeeded(cnx, messageIdData, metadata, wholePayload)) { + if (uncompressMessageIfNeeded(cnx, messageIdData, metadata, wholePayload, false)) { return Optional::of(wholePayload); } else { return Optional::empty(); @@ -430,7 +430,7 @@ void ConsumerImpl::messageReceived(const ClientConnectionPtr& cnx, const proto:: config_.getConsumerType() != ConsumerType::ConsumerShared && config_.getConsumerType() != ConsumerType::ConsumerKeyShared; if (isMessageDecryptable && !isChunkedMessage) { - if (!uncompressMessageIfNeeded(cnx, msg.message_id(), metadata, payload)) { + if (!uncompressMessageIfNeeded(cnx, msg.message_id(), metadata, payload, true)) { // Message was discarded on decompression error return; } @@ -650,7 +650,8 @@ bool ConsumerImpl::decryptMessageIfNeeded(const ClientConnectionPtr& cnx, const bool ConsumerImpl::uncompressMessageIfNeeded(const ClientConnectionPtr& cnx, const proto::MessageIdData& messageIdData, - const proto::MessageMetadata& metadata, SharedBuffer& payload) { + const proto::MessageMetadata& metadata, SharedBuffer& payload, + bool checkMaxMessageSize) { if (!metadata.has_compression()) { return true; } @@ -660,7 +661,7 @@ bool ConsumerImpl::uncompressMessageIfNeeded(const ClientConnectionPtr& cnx, uint32_t uncompressedSize = metadata.uncompressed_size(); uint32_t payloadSize = payload.readableBytes(); if (cnx) { - if (payloadSize > ClientConnection::getMaxMessageSize()) { + if (checkMaxMessageSize && payloadSize > ClientConnection::getMaxMessageSize()) { // Uncompressed size is itself corrupted since it cannot be bigger than the MaxMessageSize LOG_ERROR(getName() << "Got corrupted payload message size " << payloadSize // << " at " << messageIdData.ledgerid() << ":" << messageIdData.entryid()); diff --git a/pulsar-client-cpp/lib/ConsumerImpl.h b/pulsar-client-cpp/lib/ConsumerImpl.h index e71129a370cd9..2d4fa1eec4807 100644 --- a/pulsar-client-cpp/lib/ConsumerImpl.h +++ b/pulsar-client-cpp/lib/ConsumerImpl.h @@ -158,7 +158,8 @@ class ConsumerImpl : public ConsumerImplBase, private: bool waitingForZeroQueueSizeMessage; bool uncompressMessageIfNeeded(const ClientConnectionPtr& cnx, const proto::MessageIdData& messageIdData, - const proto::MessageMetadata& metadata, SharedBuffer& payload); + const proto::MessageMetadata& metadata, SharedBuffer& payload, + bool checkMaxMessageSize); void discardCorruptedMessage(const ClientConnectionPtr& cnx, const proto::MessageIdData& messageId, proto::CommandAck::ValidationError validationError); void increaseAvailablePermits(const ClientConnectionPtr& currentCnx, int delta = 1); diff --git a/pulsar-client-cpp/lib/ProducerImpl.cc b/pulsar-client-cpp/lib/ProducerImpl.cc index d320a2f0be922..9cbe11dbe3089 100644 --- a/pulsar-client-cpp/lib/ProducerImpl.cc +++ b/pulsar-client-cpp/lib/ProducerImpl.cc @@ -426,7 +426,7 @@ void ProducerImpl::sendAsyncWithStatsUpdate(const Message& msg, const SendCallba const bool compressed = !canAddToBatch(msg); const auto payload = - compressed ? uncompressedPayload : applyCompression(uncompressedPayload, conf_.getCompressionType()); + compressed ? applyCompression(uncompressedPayload, conf_.getCompressionType()) : uncompressedPayload; const auto compressedSize = static_cast(payload.readableBytes()); const auto maxMessageSize = static_cast(ClientConnection::getMaxMessageSize()); @@ -445,7 +445,7 @@ void ProducerImpl::sendAsyncWithStatsUpdate(const Message& msg, const SendCallba } const int totalChunks = - canAddToBatch(msg) ? 1 : getNumOfChunks(uncompressedSize, ClientConnection::getMaxMessageSize()); + canAddToBatch(msg) ? 1 : getNumOfChunks(compressedSize, ClientConnection::getMaxMessageSize()); // Each chunk should be sent individually, so try to acquire extra permits for chunks. for (int i = 0; i < (totalChunks - 1); i++) { const auto result = canEnqueueRequest(0); // size is 0 because the memory has already reserved diff --git a/pulsar-client-cpp/tests/MessageChunkingTest.cc b/pulsar-client-cpp/tests/MessageChunkingTest.cc index 15f6bf9690b6f..10f9eea7755c0 100644 --- a/pulsar-client-cpp/tests/MessageChunkingTest.cc +++ b/pulsar-client-cpp/tests/MessageChunkingTest.cc @@ -107,7 +107,9 @@ TEST_P(MessageChunkingTest, testEndToEnd) { ASSERT_EQ(msg.getMessageId(), sendMessageId); } -INSTANTIATE_TEST_SUITE_P(Pulsar, MessageChunkingTest, ::testing::Values(CompressionNone), +INSTANTIATE_TEST_SUITE_P(Pulsar, MessageChunkingTest, + ::testing::Values(CompressionNone, CompressionLZ4, CompressionZLib, CompressionZSTD, + CompressionSNAPPY), [](const ::testing::TestParamInfo& info) { return toString(info.param); }); From 007f542090bba1e3ff9532e5161c5aa55353f115 Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Wed, 5 Jan 2022 14:34:31 +0800 Subject: [PATCH 16/27] Improve logs --- pulsar-client-cpp/lib/ConsumerImpl.cc | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/pulsar-client-cpp/lib/ConsumerImpl.cc b/pulsar-client-cpp/lib/ConsumerImpl.cc index bebd92cc8e54e..67188bbf7c066 100644 --- a/pulsar-client-cpp/lib/ConsumerImpl.cc +++ b/pulsar-client-cpp/lib/ConsumerImpl.cc @@ -317,6 +317,9 @@ Optional ConsumerImpl::processMessageChunk(const SharedBuffer& pay const ClientConnectionPtr& cnx) { const auto chunkId = metadata.chunk_id(); const auto uuid = metadata.uuid(); + LOG_DEBUG("Process message chunk (chunkId: " << chunkId << ", uuid: " << uuid + << ", messageId: " << messageId << ") of " + << payload.readableBytes() << " bytes"); Lock lock(chunkProcessMutex_); auto it = chunkedMessagesMap_.find(uuid); @@ -338,12 +341,13 @@ Optional ConsumerImpl::processMessageChunk(const SharedBuffer& pay auto& chunkedMsgCtx = it->second; if (it == chunkedMessagesMap_.end() || !chunkedMsgCtx.validateChunkId(chunkId)) { if (it == chunkedMessagesMap_.end()) { - LOG_ERROR("Received unexpected chunk, messageId: " << messageId << ", chunkId: " << chunkId); + LOG_ERROR("Received an uncached chunk (uuid: " << uuid << " chunkId: " << chunkId + << ", messageId: " << messageId << ")"); } else { - LOG_ERROR("Received unexpected chunk, messageId: " << messageId << ", chunkId: " << chunkId - << ", ChunkedMessageCtx: " << chunkedMsgCtx); + LOG_ERROR("Received a chunk whose chunk id is invalid (uuid: " + << uuid << " chunkId: " << chunkId << ", messageId: " << messageId << ")"); } - chunkedMessagesMap_.erase(uuid); + removeChunkMessage(uuid, false); lock.unlock(); increaseAvailablePermits(cnx); trackMessage(messageId); From e654b6ccbb9cec985085f31ebb3504a12be51b36 Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Wed, 5 Jan 2022 16:56:08 +0800 Subject: [PATCH 17/27] Refactor chunking related fields and fix memory error --- pulsar-client-cpp/lib/ConsumerImpl.cc | 41 ++++++++++++++++----------- pulsar-client-cpp/lib/ConsumerImpl.h | 33 ++++++++++----------- 2 files changed, 41 insertions(+), 33 deletions(-) diff --git a/pulsar-client-cpp/lib/ConsumerImpl.cc b/pulsar-client-cpp/lib/ConsumerImpl.cc index 67188bbf7c066..0306b00279f44 100644 --- a/pulsar-client-cpp/lib/ConsumerImpl.cc +++ b/pulsar-client-cpp/lib/ConsumerImpl.cc @@ -326,14 +326,14 @@ Optional ConsumerImpl::processMessageChunk(const SharedBuffer& pay if (chunkId == 0) { if (it == chunkedMessagesMap_.end()) { - it = chunkedMessagesMap_ - .emplace(uuid, ChunkedMessageCtx{metadata.num_chunks_from_msg(), - metadata.total_chunk_msg_size()}) - .first; + chunkedMessagesMap_.emplace( + uuid, ChunkedMessageCtx{metadata.num_chunks_from_msg(), metadata.total_chunk_msg_size()}); + it = chunkedMessagesMap_.find(uuid); } - pendingChunkedMessage_++; - if (maxPendingChunkedMessage_ > 0 && pendingChunkedMessage_ > maxPendingChunkedMessage_) { - removeOldestPendingChunkedMessage(); + if (maxPendingChunkedMessage_ > 0 && + pendingChunkedMessageUuidQueue_.size() >= maxPendingChunkedMessage_) { + removeOldestPendingChunkedMessage(pendingChunkedMessageUuidQueue_.size() - + maxPendingChunkedMessage_ + 1); } pendingChunkedMessageUuidQueue_.emplace_back(uuid); } @@ -348,13 +348,14 @@ Optional ConsumerImpl::processMessageChunk(const SharedBuffer& pay << uuid << " chunkId: " << chunkId << ", messageId: " << messageId << ")"); } removeChunkMessage(uuid, false); + removeUuidFromQueue(uuid); lock.unlock(); increaseAvailablePermits(cnx); trackMessage(messageId); return Optional::empty(); } - chunkedMsgCtx.appendChunk(chunkId, messageId, payload); + chunkedMsgCtx.appendChunk(messageId, payload); if (!chunkedMsgCtx.isCompleted()) { lock.unlock(); increaseAvailablePermits(cnx); @@ -364,8 +365,8 @@ Optional ConsumerImpl::processMessageChunk(const SharedBuffer& pay LOG_DEBUG("Chunked message completed chunkId: " << chunkId << ", ChunkedMessageCtx: " << chunkedMsgCtx << ", sequenceId: " << metadata.sequence_id()); - removeChunkMessage(uuid, false); auto wholePayload = chunkedMsgCtx.getBuffer(); + removeChunkMessage(uuid, false); if (uncompressMessageIfNeeded(cnx, messageIdData, metadata, wholePayload, false)) { return Optional::of(wholePayload); } else { @@ -373,29 +374,35 @@ Optional ConsumerImpl::processMessageChunk(const SharedBuffer& pay } } -// It must be called when `chunkProcessMutex_` is acquired -void ConsumerImpl::removeOldestPendingChunkedMessage() { - const int numChunksToRemove = pendingChunkedMessage_ - maxPendingChunkedMessage_; +void ConsumerImpl::removeUuidFromQueue(const std::string& uuid) { + for (auto it = pendingChunkedMessageUuidQueue_.cbegin(); it != pendingChunkedMessageUuidQueue_.cend(); + ++it) { + if (*it == uuid) { + pendingChunkedMessageUuidQueue_.erase(it); + break; + } + } +} + +void ConsumerImpl::removeOldestPendingChunkedMessage(size_t numChunksToRemove) { auto it = pendingChunkedMessageUuidQueue_.begin(); - for (int i = 0; i < numChunksToRemove; i++) { + for (size_t i = 0; i < numChunksToRemove; i++) { it = pendingChunkedMessageUuidQueue_.erase(it); removeChunkMessage(*it, true); } } -// It must be called when `chunkProcessMutex_` is acquired void ConsumerImpl::removeChunkMessage(const std::string& uuid, bool processMessageId) { - pendingChunkedMessage_--; auto it = chunkedMessagesMap_.find(uuid); if (it == chunkedMessagesMap_.end()) { return; } - it = chunkedMessagesMap_.erase(it); + auto chunkedMsgCtx = std::move(it->second); + chunkedMessagesMap_.erase(it); if (!processMessageId) { return; } - const auto& chunkedMsgCtx = it->second; for (const auto& messageId : chunkedMsgCtx.getChunkedMessageIds()) { if (autoAckOldestChunkedMessageOnQueueFull_) { doAcknowledgeIndividual(messageId, [uuid, messageId](Result result) { diff --git a/pulsar-client-cpp/lib/ConsumerImpl.h b/pulsar-client-cpp/lib/ConsumerImpl.h index 2d4fa1eec4807..916eec6771cad 100644 --- a/pulsar-client-cpp/lib/ConsumerImpl.h +++ b/pulsar-client-cpp/lib/ConsumerImpl.h @@ -239,44 +239,44 @@ class ConsumerImpl : public ConsumerImplBase, ChunkedMessageCtx(const ChunkedMessageCtx&) = delete; ChunkedMessageCtx(ChunkedMessageCtx&& rhs) noexcept = default; - bool validateChunkId(int chunkId) const noexcept { - return (chunkId == lastChunkedId_ + 1) && chunkId >= 0 && chunkId < totalChunks_; - } + bool validateChunkId(int chunkId) const noexcept { return chunkId == numChunks(); } - void appendChunk(int chunkId, const MessageId& messageId, const SharedBuffer& payload) { - lastChunkedId_ = chunkId; + void appendChunk(const MessageId& messageId, const SharedBuffer& payload) { chunkedMessageIds_.emplace_back(messageId); chunkedMsgBuffer_.write(payload.data(), payload.readableBytes()); } - bool isCompleted() const noexcept { return lastChunkedId_ + 1 == totalChunks_; } + bool isCompleted() const noexcept { return totalChunks_ == numChunks(); } const SharedBuffer& getBuffer() const noexcept { return chunkedMsgBuffer_; } const std::vector& getChunkedMessageIds() const noexcept { return chunkedMessageIds_; } friend std::ostream& operator<<(std::ostream& os, const ChunkedMessageCtx& ctx) { - return os << "total chunks: " << ctx.totalChunks_ << ", last chunk id: " << ctx.lastChunkedId_; + return os << "ChunkedMessageCtx " << ctx.chunkedMsgBuffer_.readableBytes() << " of " + << ctx.chunkedMsgBuffer_.writerIndex() << " bytes, " << ctx.numChunks() << " of " + << ctx.totalChunks_ << " chunks"; } private: const int totalChunks_; SharedBuffer chunkedMsgBuffer_; std::vector chunkedMessageIds_; - int lastChunkedId_{-1}; + + int numChunks() const noexcept { return static_cast(chunkedMessageIds_.size()); } }; - mutable std::mutex chunkProcessMutex_; - std::unordered_map chunkedMessagesMap_; const size_t maxPendingChunkedMessage_; - size_t pendingChunkedMessage_{0}; - // use list here for removing uuid of expired chunks quickly - std::list pendingChunkedMessageUuidQueue_; - // if queue size is reasonable (most of the time equal to number of producers try to publish messages // concurrently on the topic) then it guards against broken chunked message which was not fully published const bool autoAckOldestChunkedMessageOnQueueFull_; + mutable std::mutex chunkProcessMutex_; + std::unordered_map chunkedMessagesMap_; + // This list contains all the keys of `chunkedMessagesMap_`. + // Here we use list for removing uuid of expired chunks quickly + std::list pendingChunkedMessageUuidQueue_; + /** * Process a chunk. If the chunk is the last chunk of a message, concatenate all buffered chunks into the * payload and return it. @@ -296,8 +296,9 @@ class ConsumerImpl : public ConsumerImplBase, const proto::MessageIdData& messageIdData, const ClientConnectionPtr& cnx); - void removeOldestPendingChunkedMessage(); - + // Following methods must be called when `chunkProcessMutex_` is acquired + void removeUuidFromQueue(const std::string& uuid); + void removeOldestPendingChunkedMessage(size_t numChunksToRemove); void removeChunkMessage(const std::string& uuid, bool processMessageId); friend class PulsarFriend; From 47b12b142252ace0e3d0502688ae2ed69cc1e300 Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Wed, 5 Jan 2022 17:24:02 +0800 Subject: [PATCH 18/27] Fix comments --- pulsar-client-cpp/lib/ConsumerImpl.h | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/pulsar-client-cpp/lib/ConsumerImpl.h b/pulsar-client-cpp/lib/ConsumerImpl.h index 916eec6771cad..334cb8cdcd297 100644 --- a/pulsar-client-cpp/lib/ConsumerImpl.h +++ b/pulsar-client-cpp/lib/ConsumerImpl.h @@ -272,9 +272,11 @@ class ConsumerImpl : public ConsumerImplBase, const bool autoAckOldestChunkedMessageOnQueueFull_; mutable std::mutex chunkProcessMutex_; + // The key is UUID, value is the associated ChunkedMessageCtx of the chunked message. std::unordered_map chunkedMessagesMap_; - // This list contains all the keys of `chunkedMessagesMap_`. - // Here we use list for removing uuid of expired chunks quickly + // This list contains all the keys of `chunkedMessagesMap_`, each key is an UUID that identifies a pending + // chunked message. Once the number of pending chunked messages exceeds the limit, the oldest UUIDs and + // the associated ChunkedMessageCtx will be removed. std::list pendingChunkedMessageUuidQueue_; /** From c56fd2b6d5c1f60895cc764319e4d8424bb37ce6 Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Wed, 5 Jan 2022 19:32:08 +0800 Subject: [PATCH 19/27] Add MapCache class --- pulsar-client-cpp/lib/MapCache.h | 117 ++++++++++++++++++++++++ pulsar-client-cpp/tests/MapCacheTest.cc | 84 +++++++++++++++++ 2 files changed, 201 insertions(+) create mode 100644 pulsar-client-cpp/lib/MapCache.h create mode 100644 pulsar-client-cpp/tests/MapCacheTest.cc diff --git a/pulsar-client-cpp/lib/MapCache.h b/pulsar-client-cpp/lib/MapCache.h new file mode 100644 index 0000000000000..37293feb69555 --- /dev/null +++ b/pulsar-client-cpp/lib/MapCache.h @@ -0,0 +1,117 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +#pragma once + +#include +#include +#include +#include + +namespace pulsar { + +// A thread safe map cache that supports removing the first N oldest entries from the map. +// Value must be moveable and have the default constructor. +template +class MapCache { + using Lock = std::lock_guard; + + mutable std::mutex mutex_; + std::unordered_map map_; + std::deque keys_; + + public: + using const_iterator = typename decltype(map_)::const_iterator; + + MapCache() = default; + MapCache(MapCache&&) noexcept = default; + + const_iterator find(const Key& key) const { + Lock lock(mutex_); + return map_.find(key); + } + + const_iterator end() const { + Lock lock(mutex_); + return map_.end(); + } + + bool putIfAbsent(const Key& key, Value&& value) { + Lock lock(mutex_); + auto it = map_.find(key); + if (it == map_.end()) { + map_.emplace(key, std::move(value)); + keys_.push_back(key); + return true; + } else { + return false; + } + } + + std::vector removeOldestValues(size_t numToRemove) { + std::vector values; + values.reserve(numToRemove); + + Lock lock(mutex_); + for (size_t i = 0; !keys_.empty() && i < numToRemove; i++) { + const auto key = keys_.front(); + auto it = map_.find(key); + if (it != map_.end()) { + values.emplace_back(std::move(it->second)); + map_.erase(it); + } + keys_.pop_front(); + } + return values; + } + + void remove(const Key& key) { + Lock lock(mutex_); + auto it = map_.find(key); + if (it != map_.end()) { + removeKeyFromKeys(key); + map_.erase(it); + } + } + + // Following methods are only used for tests + std::vector getKeys() const { + Lock lock(mutex_); + std::vector keys; + for (auto key : keys_) { + keys.emplace_back(key); + } + return keys; + } + + size_t size() const { + Lock lock(mutex_); + return map_.size(); + } + + private: + void removeKeyFromKeys(const Key& key) { + for (auto it = keys_.cbegin(); it != keys_.end(); ++it) { + if (*it == key) { + keys_.erase(it); + } + } + } +}; + +} // namespace pulsar diff --git a/pulsar-client-cpp/tests/MapCacheTest.cc b/pulsar-client-cpp/tests/MapCacheTest.cc new file mode 100644 index 0000000000000..a7810220a46b0 --- /dev/null +++ b/pulsar-client-cpp/tests/MapCacheTest.cc @@ -0,0 +1,84 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +#include +#include + +using namespace pulsar; + +struct MoveOnlyInt { + int x = 0; + + MoveOnlyInt() = default; + MoveOnlyInt(int xx) : x(xx) {} + MoveOnlyInt(const MoveOnlyInt&) = delete; + MoveOnlyInt(MoveOnlyInt&& rhs) noexcept : x(rhs.x) {} + + bool operator=(const MoveOnlyInt& rhs) const { return x == rhs.x; } +}; + +using VecInt = std::vector; + +inline VecInt toIntVec(const std::vector& v) { + VecInt result; + for (const auto& i : v) { + result.emplace_back(i.x); + } + return result; +} + +TEST(MapCacheTest, testPutIfAbsent) { + MapCache cache; + + ASSERT_TRUE(cache.putIfAbsent(1, {100})); + ASSERT_FALSE(cache.putIfAbsent(1, {200})); + auto it = cache.find(1); + ASSERT_NE(it, cache.end()); + ASSERT_EQ(it->second.x, 100); + + cache.remove(1); + ASSERT_EQ(cache.find(1), cache.end()); +} + +TEST(MapCacheTest, testRemoveOldestValues) { + MapCache cache; + ASSERT_TRUE(cache.putIfAbsent(1, {200})); + ASSERT_TRUE(cache.putIfAbsent(2, {210})); + ASSERT_TRUE(cache.putIfAbsent(3, {220})); + ASSERT_EQ(cache.getKeys(), (VecInt{1, 2, 3})); + + ASSERT_EQ(toIntVec(cache.removeOldestValues(2)), (VecInt{200, 210})); + + ASSERT_EQ(cache.getKeys(), (VecInt{3})); + ASSERT_EQ(cache.size(), 1); + auto it = cache.find(3); + ASSERT_NE(it, cache.end()); + ASSERT_EQ(it->second.x, 220); +} + +TEST(MapCacheTest, testRemoveAllValues) { + MapCache cache; + ASSERT_TRUE(cache.putIfAbsent(1, {300})); + ASSERT_TRUE(cache.putIfAbsent(2, {310})); + ASSERT_TRUE(cache.putIfAbsent(3, {320})); + + // removeOldestValues works well even if the argument is greater than the size of keys + ASSERT_EQ(toIntVec(cache.removeOldestValues(10000)), (VecInt{300, 310, 320})); + ASSERT_TRUE(cache.getKeys().empty()); + ASSERT_EQ(cache.size(), 0); +} From ba92b2afc75a4508a346d41d078d645f94aa57b1 Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Wed, 5 Jan 2022 20:25:35 +0800 Subject: [PATCH 20/27] Use MapCache to refactor ConsumerImpl --- pulsar-client-cpp/lib/ConsumerImpl.cc | 78 +++++-------------- pulsar-client-cpp/lib/ConsumerImpl.h | 11 ++- pulsar-client-cpp/lib/MapCache.h | 49 ++++-------- pulsar-client-cpp/tests/MapCacheTest.cc | 38 ++++----- .../tests/MessageChunkingTest.cc | 45 +++++++---- 5 files changed, 91 insertions(+), 130 deletions(-) diff --git a/pulsar-client-cpp/lib/ConsumerImpl.cc b/pulsar-client-cpp/lib/ConsumerImpl.cc index 0306b00279f44..69e8cfedf126d 100644 --- a/pulsar-client-cpp/lib/ConsumerImpl.cc +++ b/pulsar-client-cpp/lib/ConsumerImpl.cc @@ -322,33 +322,40 @@ Optional ConsumerImpl::processMessageChunk(const SharedBuffer& pay << payload.readableBytes() << " bytes"); Lock lock(chunkProcessMutex_); - auto it = chunkedMessagesMap_.find(uuid); + auto it = chunkedMessageCache_.find(uuid); if (chunkId == 0) { - if (it == chunkedMessagesMap_.end()) { - chunkedMessagesMap_.emplace( + if (it == chunkedMessageCache_.end()) { + it = chunkedMessageCache_.putIfAbsent( uuid, ChunkedMessageCtx{metadata.num_chunks_from_msg(), metadata.total_chunk_msg_size()}); - it = chunkedMessagesMap_.find(uuid); } - if (maxPendingChunkedMessage_ > 0 && - pendingChunkedMessageUuidQueue_.size() >= maxPendingChunkedMessage_) { - removeOldestPendingChunkedMessage(pendingChunkedMessageUuidQueue_.size() - - maxPendingChunkedMessage_ + 1); + if (maxPendingChunkedMessage_ > 0 && chunkedMessageCache_.size() >= maxPendingChunkedMessage_) { + chunkedMessageCache_.removeOldestValues( + chunkedMessageCache_.size() - maxPendingChunkedMessage_ + 1, + [this, messageId](const std::string& uuid, const ChunkedMessageCtx& ctx) { + if (autoAckOldestChunkedMessageOnQueueFull_) { + doAcknowledgeIndividual(messageId, [uuid, messageId](Result result) { + if (result != ResultOk) { + LOG_WARN("Failed to acknowledge discarded chunk, uuid: " + << uuid << ", messageId: " << messageId); + } + }); + } else { + trackMessage(messageId); + } + }); } - pendingChunkedMessageUuidQueue_.emplace_back(uuid); } auto& chunkedMsgCtx = it->second; - if (it == chunkedMessagesMap_.end() || !chunkedMsgCtx.validateChunkId(chunkId)) { - if (it == chunkedMessagesMap_.end()) { + if (it == chunkedMessageCache_.end() || !chunkedMsgCtx.validateChunkId(chunkId)) { + if (it == chunkedMessageCache_.end()) { LOG_ERROR("Received an uncached chunk (uuid: " << uuid << " chunkId: " << chunkId << ", messageId: " << messageId << ")"); } else { LOG_ERROR("Received a chunk whose chunk id is invalid (uuid: " << uuid << " chunkId: " << chunkId << ", messageId: " << messageId << ")"); } - removeChunkMessage(uuid, false); - removeUuidFromQueue(uuid); lock.unlock(); increaseAvailablePermits(cnx); trackMessage(messageId); @@ -366,7 +373,7 @@ Optional ConsumerImpl::processMessageChunk(const SharedBuffer& pay << ", sequenceId: " << metadata.sequence_id()); auto wholePayload = chunkedMsgCtx.getBuffer(); - removeChunkMessage(uuid, false); + chunkedMessageCache_.remove(uuid); if (uncompressMessageIfNeeded(cnx, messageIdData, metadata, wholePayload, false)) { return Optional::of(wholePayload); } else { @@ -374,49 +381,6 @@ Optional ConsumerImpl::processMessageChunk(const SharedBuffer& pay } } -void ConsumerImpl::removeUuidFromQueue(const std::string& uuid) { - for (auto it = pendingChunkedMessageUuidQueue_.cbegin(); it != pendingChunkedMessageUuidQueue_.cend(); - ++it) { - if (*it == uuid) { - pendingChunkedMessageUuidQueue_.erase(it); - break; - } - } -} - -void ConsumerImpl::removeOldestPendingChunkedMessage(size_t numChunksToRemove) { - auto it = pendingChunkedMessageUuidQueue_.begin(); - for (size_t i = 0; i < numChunksToRemove; i++) { - it = pendingChunkedMessageUuidQueue_.erase(it); - removeChunkMessage(*it, true); - } -} - -void ConsumerImpl::removeChunkMessage(const std::string& uuid, bool processMessageId) { - auto it = chunkedMessagesMap_.find(uuid); - if (it == chunkedMessagesMap_.end()) { - return; - } - auto chunkedMsgCtx = std::move(it->second); - chunkedMessagesMap_.erase(it); - - if (!processMessageId) { - return; - } - for (const auto& messageId : chunkedMsgCtx.getChunkedMessageIds()) { - if (autoAckOldestChunkedMessageOnQueueFull_) { - doAcknowledgeIndividual(messageId, [uuid, messageId](Result result) { - if (result != ResultOk) { - LOG_WARN("Failed to acknowledge discarded chunk, uuid: " << uuid - << ", messageId: " << messageId); - } - }); - } else { - trackMessage(messageId); - } - } -} - void ConsumerImpl::messageReceived(const ClientConnectionPtr& cnx, const proto::CommandMessage& msg, bool& isChecksumValid, proto::MessageMetadata& metadata, SharedBuffer& payload) { diff --git a/pulsar-client-cpp/lib/ConsumerImpl.h b/pulsar-client-cpp/lib/ConsumerImpl.h index 334cb8cdcd297..4cd803d7aa397 100644 --- a/pulsar-client-cpp/lib/ConsumerImpl.h +++ b/pulsar-client-cpp/lib/ConsumerImpl.h @@ -40,6 +40,7 @@ #include "BatchAcknowledgementTracker.h" #include #include +#include #include #include #include @@ -271,7 +272,6 @@ class ConsumerImpl : public ConsumerImplBase, // concurrently on the topic) then it guards against broken chunked message which was not fully published const bool autoAckOldestChunkedMessageOnQueueFull_; - mutable std::mutex chunkProcessMutex_; // The key is UUID, value is the associated ChunkedMessageCtx of the chunked message. std::unordered_map chunkedMessagesMap_; // This list contains all the keys of `chunkedMessagesMap_`, each key is an UUID that identifies a pending @@ -279,6 +279,10 @@ class ConsumerImpl : public ConsumerImplBase, // the associated ChunkedMessageCtx will be removed. std::list pendingChunkedMessageUuidQueue_; + // The key is UUID, value is the associated ChunkedMessageCtx of the chunked message. + MapCache chunkedMessageCache_; + mutable std::mutex chunkProcessMutex_; + /** * Process a chunk. If the chunk is the last chunk of a message, concatenate all buffered chunks into the * payload and return it. @@ -298,11 +302,6 @@ class ConsumerImpl : public ConsumerImplBase, const proto::MessageIdData& messageIdData, const ClientConnectionPtr& cnx); - // Following methods must be called when `chunkProcessMutex_` is acquired - void removeUuidFromQueue(const std::string& uuid); - void removeOldestPendingChunkedMessage(size_t numChunksToRemove); - void removeChunkMessage(const std::string& uuid, bool processMessageId); - friend class PulsarFriend; // these two declared friend to access setNegativeAcknowledgeEnabledForTesting diff --git a/pulsar-client-cpp/lib/MapCache.h b/pulsar-client-cpp/lib/MapCache.h index 37293feb69555..a3a4fb0e065e8 100644 --- a/pulsar-client-cpp/lib/MapCache.h +++ b/pulsar-client-cpp/lib/MapCache.h @@ -19,69 +19,60 @@ #pragma once #include -#include +#include #include #include namespace pulsar { -// A thread safe map cache that supports removing the first N oldest entries from the map. +// A map cache that supports removing the first N oldest entries from the map. // Value must be moveable and have the default constructor. template class MapCache { - using Lock = std::lock_guard; - - mutable std::mutex mutex_; std::unordered_map map_; std::deque keys_; public: using const_iterator = typename decltype(map_)::const_iterator; + using iterator = typename decltype(map_)::iterator; MapCache() = default; MapCache(MapCache&&) noexcept = default; - const_iterator find(const Key& key) const { - Lock lock(mutex_); - return map_.find(key); - } + size_t size() const noexcept { return map_.size(); } - const_iterator end() const { - Lock lock(mutex_); - return map_.end(); - } + const_iterator find(const Key& key) const { return map_.find(key); } + iterator find(const Key& key) { return map_.find(key); } + + const_iterator end() const noexcept { return map_.end(); } + iterator end() noexcept { return map_.end(); } - bool putIfAbsent(const Key& key, Value&& value) { - Lock lock(mutex_); + iterator putIfAbsent(const Key& key, Value&& value) { auto it = map_.find(key); if (it == map_.end()) { - map_.emplace(key, std::move(value)); keys_.push_back(key); - return true; + return map_.emplace(key, std::move(value)).first; } else { - return false; + return end(); } } - std::vector removeOldestValues(size_t numToRemove) { - std::vector values; - values.reserve(numToRemove); - - Lock lock(mutex_); + void removeOldestValues(size_t numToRemove, + const std::function& callback) { for (size_t i = 0; !keys_.empty() && i < numToRemove; i++) { const auto key = keys_.front(); auto it = map_.find(key); if (it != map_.end()) { - values.emplace_back(std::move(it->second)); + if (callback) { + callback(it->first, it->second); + } map_.erase(it); } keys_.pop_front(); } - return values; } void remove(const Key& key) { - Lock lock(mutex_); auto it = map_.find(key); if (it != map_.end()) { removeKeyFromKeys(key); @@ -91,7 +82,6 @@ class MapCache { // Following methods are only used for tests std::vector getKeys() const { - Lock lock(mutex_); std::vector keys; for (auto key : keys_) { keys.emplace_back(key); @@ -99,11 +89,6 @@ class MapCache { return keys; } - size_t size() const { - Lock lock(mutex_); - return map_.size(); - } - private: void removeKeyFromKeys(const Key& key) { for (auto it = keys_.cbegin(); it != keys_.end(); ++it) { diff --git a/pulsar-client-cpp/tests/MapCacheTest.cc b/pulsar-client-cpp/tests/MapCacheTest.cc index a7810220a46b0..12a89ee17be34 100644 --- a/pulsar-client-cpp/tests/MapCacheTest.cc +++ b/pulsar-client-cpp/tests/MapCacheTest.cc @@ -32,21 +32,11 @@ struct MoveOnlyInt { bool operator=(const MoveOnlyInt& rhs) const { return x == rhs.x; } }; -using VecInt = std::vector; - -inline VecInt toIntVec(const std::vector& v) { - VecInt result; - for (const auto& i : v) { - result.emplace_back(i.x); - } - return result; -} - TEST(MapCacheTest, testPutIfAbsent) { MapCache cache; - ASSERT_TRUE(cache.putIfAbsent(1, {100})); - ASSERT_FALSE(cache.putIfAbsent(1, {200})); + ASSERT_NE(cache.putIfAbsent(1, {100}), cache.end()); + ASSERT_EQ(cache.putIfAbsent(1, {200}), cache.end()); auto it = cache.find(1); ASSERT_NE(it, cache.end()); ASSERT_EQ(it->second.x, 100); @@ -57,14 +47,18 @@ TEST(MapCacheTest, testPutIfAbsent) { TEST(MapCacheTest, testRemoveOldestValues) { MapCache cache; - ASSERT_TRUE(cache.putIfAbsent(1, {200})); - ASSERT_TRUE(cache.putIfAbsent(2, {210})); - ASSERT_TRUE(cache.putIfAbsent(3, {220})); - ASSERT_EQ(cache.getKeys(), (VecInt{1, 2, 3})); + cache.putIfAbsent(1, {200}); + cache.putIfAbsent(2, {210}); + cache.putIfAbsent(3, {220}); + ASSERT_EQ(cache.getKeys(), (std::vector{1, 2, 3})); - ASSERT_EQ(toIntVec(cache.removeOldestValues(2)), (VecInt{200, 210})); + std::vector removedValues; + cache.removeOldestValues(2, [&removedValues](const int& key, const MoveOnlyInt& value) { + removedValues.emplace_back(value.x); + }); + ASSERT_EQ(removedValues, (std::vector{200, 210})); - ASSERT_EQ(cache.getKeys(), (VecInt{3})); + ASSERT_EQ(cache.getKeys(), (std::vector{3})); ASSERT_EQ(cache.size(), 1); auto it = cache.find(3); ASSERT_NE(it, cache.end()); @@ -73,12 +67,12 @@ TEST(MapCacheTest, testRemoveOldestValues) { TEST(MapCacheTest, testRemoveAllValues) { MapCache cache; - ASSERT_TRUE(cache.putIfAbsent(1, {300})); - ASSERT_TRUE(cache.putIfAbsent(2, {310})); - ASSERT_TRUE(cache.putIfAbsent(3, {320})); + cache.putIfAbsent(1, {300}); + cache.putIfAbsent(2, {310}); + cache.putIfAbsent(3, {320}); // removeOldestValues works well even if the argument is greater than the size of keys - ASSERT_EQ(toIntVec(cache.removeOldestValues(10000)), (VecInt{300, 310, 320})); + cache.removeOldestValues(10000, nullptr); ASSERT_TRUE(cache.getKeys().empty()); ASSERT_EQ(cache.size(), 0); } diff --git a/pulsar-client-cpp/tests/MessageChunkingTest.cc b/pulsar-client-cpp/tests/MessageChunkingTest.cc index 10f9eea7755c0..66c9ea7700208 100644 --- a/pulsar-client-cpp/tests/MessageChunkingTest.cc +++ b/pulsar-client-cpp/tests/MessageChunkingTest.cc @@ -49,8 +49,20 @@ static std::string toString(CompressionType compressionType) { } } +inline std::string createLargeMessage() { + std::string largeMessage(maxMessageSize * 3, 'a'); + std::default_random_engine e(time(nullptr)); + std::uniform_int_distribution u(0, 25); + for (size_t i = 0; i < largeMessage.size(); i++) { + largeMessage[i] = 'a' + u(e); + } + return largeMessage; +} + class MessageChunkingTest : public ::testing::TestWithParam { public: + static std::string largeMessage; + void TearDown() override { client_.close(); } void createProducer(const std::string& topic, Producer& producer) { @@ -71,6 +83,8 @@ class MessageChunkingTest : public ::testing::TestWithParam { Client client_{lookupUrl}; }; +std::string MessageChunkingTest::largeMessage = createLargeMessage(); + TEST_F(MessageChunkingTest, testInvalidConfig) { Client client(lookupUrl); ProducerConfiguration conf; @@ -89,22 +103,27 @@ TEST_P(MessageChunkingTest, testEndToEnd) { Producer producer; createProducer(topic, producer); - std::string largeMessage(maxMessageSize * 3, 'a'); - std::default_random_engine e(time(nullptr)); - std::uniform_int_distribution u(0, 25); - for (auto& ch : largeMessage) { - ch = 'a' + u(e); - } + constexpr int numMessages = 10; - MessageId sendMessageId; - ASSERT_EQ(ResultOk, producer.send(MessageBuilder().setContent(largeMessage).build(), sendMessageId)); - LOG_INFO("Send to " << sendMessageId); + std::vector sendMessageIds; + for (int i = 0; i < numMessages; i++) { + MessageId messageId; + ASSERT_EQ(ResultOk, producer.send(MessageBuilder().setContent(largeMessage).build(), messageId)); + LOG_INFO("Send " << i << " to " << messageId); + sendMessageIds.emplace_back(messageId); + } Message msg; - ASSERT_EQ(ResultOk, consumer.receive(msg, 3000)); - LOG_INFO("Receive " << msg.getLength() << " bytes from " << msg.getMessageId()); - ASSERT_EQ(msg.getDataAsString(), largeMessage); - ASSERT_EQ(msg.getMessageId(), sendMessageId); + std::vector receivedMessageIds; + for (int i = 0; i < numMessages; i++) { + ASSERT_EQ(ResultOk, consumer.receive(msg, 3000)); + LOG_INFO("Receive " << msg.getLength() << " bytes from " << msg.getMessageId()); + ASSERT_EQ(msg.getDataAsString(), largeMessage); + receivedMessageIds.emplace_back(msg.getMessageId()); + } + ASSERT_EQ(receivedMessageIds, sendMessageIds); + ASSERT_EQ(receivedMessageIds.front().ledgerId(), receivedMessageIds.front().ledgerId()); + ASSERT_GT(receivedMessageIds.back().entryId(), numMessages); } INSTANTIATE_TEST_SUITE_P(Pulsar, MessageChunkingTest, From 11a13a808c4a7e3ecd0312081c7453e9f6028e20 Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Wed, 5 Jan 2022 20:31:10 +0800 Subject: [PATCH 21/27] Verify the chunk cache is cleared --- pulsar-client-cpp/tests/MessageChunkingTest.cc | 5 +++++ pulsar-client-cpp/tests/PulsarFriend.h | 6 ++++++ 2 files changed, 11 insertions(+) diff --git a/pulsar-client-cpp/tests/MessageChunkingTest.cc b/pulsar-client-cpp/tests/MessageChunkingTest.cc index 66c9ea7700208..3b106bc9b110f 100644 --- a/pulsar-client-cpp/tests/MessageChunkingTest.cc +++ b/pulsar-client-cpp/tests/MessageChunkingTest.cc @@ -22,6 +22,7 @@ #include #include #include "lib/LogUtils.h" +#include "PulsarFriend.h" DECLARE_LOG_OBJECT() @@ -124,6 +125,10 @@ TEST_P(MessageChunkingTest, testEndToEnd) { ASSERT_EQ(receivedMessageIds, sendMessageIds); ASSERT_EQ(receivedMessageIds.front().ledgerId(), receivedMessageIds.front().ledgerId()); ASSERT_GT(receivedMessageIds.back().entryId(), numMessages); + + // Verify the cache has been cleared + auto& chunkedMessageCache = PulsarFriend::getChunkedMessageCache(consumer); + ASSERT_EQ(chunkedMessageCache.size(), 0); } INSTANTIATE_TEST_SUITE_P(Pulsar, MessageChunkingTest, diff --git a/pulsar-client-cpp/tests/PulsarFriend.h b/pulsar-client-cpp/tests/PulsarFriend.h index aed7096366ad8..c7aa3a67bb957 100644 --- a/pulsar-client-cpp/tests/PulsarFriend.h +++ b/pulsar-client-cpp/tests/PulsarFriend.h @@ -79,6 +79,12 @@ class PulsarFriend { return std::static_pointer_cast(consumer.impl_); } + static decltype(ConsumerImpl::chunkedMessageCache_) & getChunkedMessageCache(Consumer consumer) { + auto consumerImpl = getConsumerImplPtr(consumer); + ConsumerImpl::Lock lock(consumerImpl->chunkProcessMutex_); + return consumerImpl->chunkedMessageCache_; + } + static std::shared_ptr getPartitionedConsumerImplPtr(Consumer consumer) { return std::static_pointer_cast(consumer.impl_); } From 58c144dd6e98cf89758303863b1d3e79295088b8 Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Wed, 5 Jan 2022 21:23:06 +0800 Subject: [PATCH 22/27] Fix chunked cache --- pulsar-client-cpp/lib/ConsumerImpl.cc | 3 +++ 1 file changed, 3 insertions(+) diff --git a/pulsar-client-cpp/lib/ConsumerImpl.cc b/pulsar-client-cpp/lib/ConsumerImpl.cc index 69e8cfedf126d..4d4a135dbec69 100644 --- a/pulsar-client-cpp/lib/ConsumerImpl.cc +++ b/pulsar-client-cpp/lib/ConsumerImpl.cc @@ -344,6 +344,8 @@ Optional ConsumerImpl::processMessageChunk(const SharedBuffer& pay trackMessage(messageId); } }); + it = chunkedMessageCache_.putIfAbsent( + uuid, ChunkedMessageCtx{metadata.num_chunks_from_msg(), metadata.total_chunk_msg_size()}); } } @@ -355,6 +357,7 @@ Optional ConsumerImpl::processMessageChunk(const SharedBuffer& pay } else { LOG_ERROR("Received a chunk whose chunk id is invalid (uuid: " << uuid << " chunkId: " << chunkId << ", messageId: " << messageId << ")"); + chunkedMessageCache_.remove(uuid); } lock.unlock(); increaseAvailablePermits(cnx); From e8933005694d78f309cbf0f47cbf777102c0e996 Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Wed, 5 Jan 2022 23:16:47 +0800 Subject: [PATCH 23/27] Fix CentOS 7 build --- pulsar-client-cpp/lib/ConsumerImpl.h | 6 +++++- pulsar-client-cpp/lib/MapCache.h | 5 +++-- 2 files changed, 8 insertions(+), 3 deletions(-) diff --git a/pulsar-client-cpp/lib/ConsumerImpl.h b/pulsar-client-cpp/lib/ConsumerImpl.h index 4cd803d7aa397..2bdb82fc7630a 100644 --- a/pulsar-client-cpp/lib/ConsumerImpl.h +++ b/pulsar-client-cpp/lib/ConsumerImpl.h @@ -238,7 +238,11 @@ class ConsumerImpl : public ConsumerImplBase, } ChunkedMessageCtx(const ChunkedMessageCtx&) = delete; - ChunkedMessageCtx(ChunkedMessageCtx&& rhs) noexcept = default; + // Here we don't use =default to be compatible with GCC 4.8 + ChunkedMessageCtx(ChunkedMessageCtx&& rhs) noexcept + : totalChunks_(rhs.totalChunks_), + chunkedMsgBuffer_(std::move(rhs.chunkedMsgBuffer_)), + chunkedMessageIds_(std::move(rhs.chunkedMessageIds_)) {} bool validateChunkId(int chunkId) const noexcept { return chunkId == numChunks(); } diff --git a/pulsar-client-cpp/lib/MapCache.h b/pulsar-client-cpp/lib/MapCache.h index a3a4fb0e065e8..d8b28a00b6b13 100644 --- a/pulsar-client-cpp/lib/MapCache.h +++ b/pulsar-client-cpp/lib/MapCache.h @@ -37,7 +37,8 @@ class MapCache { using iterator = typename decltype(map_)::iterator; MapCache() = default; - MapCache(MapCache&&) noexcept = default; + // Here we don't use =default to be compatible with GCC 4.8 + MapCache(MapCache&& rhs) noexcept : map_(std::move(rhs.map_)), keys_(std::move(rhs.keys_)) {} size_t size() const noexcept { return map_.size(); } @@ -91,7 +92,7 @@ class MapCache { private: void removeKeyFromKeys(const Key& key) { - for (auto it = keys_.cbegin(); it != keys_.end(); ++it) { + for (auto it = keys_.begin(); it != keys_.end(); ++it) { if (*it == key) { keys_.erase(it); } From 1d91fb0f9c67eb405f15528547458567808e6a28 Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Wed, 5 Jan 2022 23:56:20 +0800 Subject: [PATCH 24/27] Fix Ubuntu 16.04 build failure --- pulsar-client-cpp/tests/MessageChunkingTest.cc | 13 +++++++------ 1 file changed, 7 insertions(+), 6 deletions(-) diff --git a/pulsar-client-cpp/tests/MessageChunkingTest.cc b/pulsar-client-cpp/tests/MessageChunkingTest.cc index 3b106bc9b110f..e4412dee8d8d4 100644 --- a/pulsar-client-cpp/tests/MessageChunkingTest.cc +++ b/pulsar-client-cpp/tests/MessageChunkingTest.cc @@ -131,9 +131,10 @@ TEST_P(MessageChunkingTest, testEndToEnd) { ASSERT_EQ(chunkedMessageCache.size(), 0); } -INSTANTIATE_TEST_SUITE_P(Pulsar, MessageChunkingTest, - ::testing::Values(CompressionNone, CompressionLZ4, CompressionZLib, CompressionZSTD, - CompressionSNAPPY), - [](const ::testing::TestParamInfo& info) { - return toString(info.param); - }); +// The CI env is Ubuntu 16.04, the gtest-dev version is 1.8.0 that doesn't have INSTANTIATE_TEST_SUITE_P +INSTANTIATE_TEST_CASE_P(Pulsar, MessageChunkingTest, + ::testing::Values(CompressionNone, CompressionLZ4, CompressionZLib, CompressionZSTD, + CompressionSNAPPY), + [](const ::testing::TestParamInfo& info) { + return toString(info.param); + }); From def6a4d1eb918367bc4ce277266c84b66be24c18 Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Thu, 6 Jan 2022 00:25:12 +0800 Subject: [PATCH 25/27] Fix incompatibility with GTest 1.8.0 --- pulsar-client-cpp/tests/MessageChunkingTest.cc | 5 +---- 1 file changed, 1 insertion(+), 4 deletions(-) diff --git a/pulsar-client-cpp/tests/MessageChunkingTest.cc b/pulsar-client-cpp/tests/MessageChunkingTest.cc index e4412dee8d8d4..db81efdc3b154 100644 --- a/pulsar-client-cpp/tests/MessageChunkingTest.cc +++ b/pulsar-client-cpp/tests/MessageChunkingTest.cc @@ -134,7 +134,4 @@ TEST_P(MessageChunkingTest, testEndToEnd) { // The CI env is Ubuntu 16.04, the gtest-dev version is 1.8.0 that doesn't have INSTANTIATE_TEST_SUITE_P INSTANTIATE_TEST_CASE_P(Pulsar, MessageChunkingTest, ::testing::Values(CompressionNone, CompressionLZ4, CompressionZLib, CompressionZSTD, - CompressionSNAPPY), - [](const ::testing::TestParamInfo& info) { - return toString(info.param); - }); + CompressionSNAPPY)); From acaef757eee54f5a6763fddfe828a3b69eb0ac25 Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Thu, 6 Jan 2022 12:36:56 +0800 Subject: [PATCH 26/27] Fix tests --- pulsar-client-cpp/lib/ProducerImpl.cc | 6 ++++-- pulsar-client-cpp/test-conf/standalone-ssl.conf | 2 +- pulsar-client-cpp/tests/BasicEndToEndTest.cc | 8 ++++---- pulsar-client-cpp/tests/MessageChunkingTest.cc | 2 +- 4 files changed, 10 insertions(+), 8 deletions(-) diff --git a/pulsar-client-cpp/lib/ProducerImpl.cc b/pulsar-client-cpp/lib/ProducerImpl.cc index 9cbe11dbe3089..87f8552ee4516 100644 --- a/pulsar-client-cpp/lib/ProducerImpl.cc +++ b/pulsar-client-cpp/lib/ProducerImpl.cc @@ -391,7 +391,9 @@ void ProducerImpl::sendAsync(const Message& msg, SendCallback callback) { auto self = shared_from_this(); sendAsyncWithStatsUpdate(msg, [this, self, now, callback](Result result, const MessageId& messageId) { producerStatsBasePtr_->messageReceived(result, now); - callback(result, messageId); + if (callback) { + callback(result, messageId); + } }); } @@ -430,7 +432,7 @@ void ProducerImpl::sendAsyncWithStatsUpdate(const Message& msg, const SendCallba const auto compressedSize = static_cast(payload.readableBytes()); const auto maxMessageSize = static_cast(ClientConnection::getMaxMessageSize()); - if (compressedSize > ClientConnection::getMaxMessageSize() && !chunkingEnabled_) { + if (compressed && compressedSize > ClientConnection::getMaxMessageSize() && !chunkingEnabled_) { LOG_WARN(getName() << " - compressed Message payload size " << payload.readableBytes() << " cannot exceed " << ClientConnection::getMaxMessageSize() << " bytes unless chunking is enabled"); diff --git a/pulsar-client-cpp/test-conf/standalone-ssl.conf b/pulsar-client-cpp/test-conf/standalone-ssl.conf index 8fedee7054be1..5c94e58720deb 100644 --- a/pulsar-client-cpp/test-conf/standalone-ssl.conf +++ b/pulsar-client-cpp/test-conf/standalone-ssl.conf @@ -303,4 +303,4 @@ globalZookeeperServers={{ zookeeper_servers }} brokerServicePurgeInactiveFrequencyInSeconds=60 # Given a specific limit of the max message size -maxMessageSize=10240 +maxMessageSize=1024000 diff --git a/pulsar-client-cpp/tests/BasicEndToEndTest.cc b/pulsar-client-cpp/tests/BasicEndToEndTest.cc index be6577523ed7d..da5c60952dd18 100644 --- a/pulsar-client-cpp/tests/BasicEndToEndTest.cc +++ b/pulsar-client-cpp/tests/BasicEndToEndTest.cc @@ -602,7 +602,7 @@ TEST(BasicEndToEndTest, testMessageTooBig) { Result result = client.createProducer(topicName, conf, producer); ASSERT_EQ(ResultOk, result); - int size = Commands::DefaultMaxMessageSize + 1000 * 100; + int size = ClientConnection::getMaxMessageSize() + 1000 * 100; char *content = new char[size]; memset(content, 0, size); Message msg = MessageBuilder().setAllocatedContent(content, size).build(); @@ -610,7 +610,7 @@ TEST(BasicEndToEndTest, testMessageTooBig) { ASSERT_EQ(ResultMessageTooBig, result); // Anything up to MaxMessageSize should be allowed - size = Commands::DefaultMaxMessageSize; + size = ClientConnection::getMaxMessageSize(); msg = MessageBuilder().setAllocatedContent(content, size).build(); result = producer.send(msg); ASSERT_EQ(ResultOk, result); @@ -1156,7 +1156,7 @@ TEST(BasicEndToEndTest, testProduceMessageSize) { result = producerFuture.get(producer2); ASSERT_EQ(ResultOk, result); - int size = Commands::DefaultMaxMessageSize + 1000 * 100; + int size = ClientConnection::getMaxMessageSize() + 1000 * 100; char *content = new char[size]; memset(content, 0, size); Message msg = MessageBuilder().setAllocatedContent(content, size).build(); @@ -1208,7 +1208,7 @@ TEST(BasicEndToEndTest, testBigMessageSizeBatching) { result = client.createProducer(topicName, conf2, producer2); ASSERT_EQ(ResultOk, result); - int size = Commands::DefaultMaxMessageSize + 1000 * 100; + int size = ClientConnection::getMaxMessageSize() + 1000 * 100; char *content = new char[size]; memset(content, 0, size); Message msg = MessageBuilder().setAllocatedContent(content, size).build(); diff --git a/pulsar-client-cpp/tests/MessageChunkingTest.cc b/pulsar-client-cpp/tests/MessageChunkingTest.cc index db81efdc3b154..ae0114cefb162 100644 --- a/pulsar-client-cpp/tests/MessageChunkingTest.cc +++ b/pulsar-client-cpp/tests/MessageChunkingTest.cc @@ -31,7 +31,7 @@ using namespace pulsar; static const std::string lookupUrl = "pulsar://localhost:6650"; // See the `maxMessageSize` config in test-conf/standalone-ssl.conf -static constexpr size_t maxMessageSize = 10240; +static constexpr size_t maxMessageSize = 1024000; static std::string toString(CompressionType compressionType) { switch (compressionType) { From 57169610903d9f23fa52965979feb205947fd8f2 Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Thu, 6 Jan 2022 13:55:40 +0800 Subject: [PATCH 27/27] Fix GCC 5.4 segmentation fault --- pulsar-client-cpp/lib/MapCache.h | 1 + 1 file changed, 1 insertion(+) diff --git a/pulsar-client-cpp/lib/MapCache.h b/pulsar-client-cpp/lib/MapCache.h index d8b28a00b6b13..b9a0069eaf44a 100644 --- a/pulsar-client-cpp/lib/MapCache.h +++ b/pulsar-client-cpp/lib/MapCache.h @@ -95,6 +95,7 @@ class MapCache { for (auto it = keys_.begin(); it != keys_.end(); ++it) { if (*it == key) { keys_.erase(it); + break; } } }