CPP add receiveAsync API

[rdhabalia]

Motivation

Sometimes, client wants to receive a message async to complete specific depending task or request. eg. java-client library has receiveAsync api which is used by proxy to receive msg async and completes pending proxy request on message-received. CPP-client library doesn't have receiveAsync api.

Modifications

Add receiveAsync api to cpp-client library.

[jai1]

Why remove this mutex ?

[rdhabalia]

Do you see any use of this mutex_ here?
in fact I think it can create a dead-lock.
eg: if incomingMessages-queue is empty then receive() will take the mutex and wait on it until message is received. Now, when message is received and if it tries to take mutex_ then it will create a deadlock.
However, for batch msg we are using pendingReceiveMutex_ into receiveIndividualMessagesFromBatch to sync msg-queue and pending-receive queue.

[jai1]

A. Yes I need the mutex since batchAcknowledgementTracker_ is not thread safe
B. As discussed - I don't see a deadlock

[jai1]

Receive Callback should return a Result along with Message

[rdhabalia]

it's a async-receive and it will never have any failure as a result and it only returns a message when available so, I think there is no need of result in this case.??

[jai1]

What if consumer is not initialized - most of the other async operations in Consumer.cc return a ResultConsumerNotInitialized .

[rdhabalia]

then we should add request into pending-queue and it should complete the callback when the message will be available.

[rdhabalia]

however, we should immediately fail the callback if consumer is closing or already closed (I will make similar change in java-client as well). so, I have added the result into the callback.

[jai1]

UnboundedBlockingQueue is not a good idea, since you don't want to block if the queue is empty
I would suggest using a vector or a list.

[rdhabalia]

Sure. actually, my motivation to use UnboundedBlockingQueue to have thread-safe, unbounded collection. Do you see any specific issue with UnboundedBlockingQueue? Also I think std::list/queue creates a copy of object on pop_front()/front()?

[rdhabalia]

replaced UnboundedBlockingQueue with std::queue

[jai1]

No need for a separate pendingReceiveMutex_ - I suggest to use mutex_

[rdhabalia]

pendingReceiveMutex_ is a mutex which requires to sync pendingReceives-queue and incomingMessages-queue together to make sure message-queue is empty while adding into pending-queue and also requires to prevent race-condition on pendingReceive_ .
mutex_ will not work here for example:

• receive() is a blocking call which takes locks on mutext_ until message is received
• now, if we use the same mutex_ on message-received then it will create a deadlock.

[jai1]

Agreed, but the problem is that in PartitionedConsumerImpl:receive we should have released the lock after we checked the state_ and messageListener_ condition since the other data structures are thread safe (see ConsumerImpl::receiveHelper to understand my point)

If you fix this locking then you won't require pendingReceiveMutex_

I am stressing about this minor thing because in all classes we have only one lock to protect all members (mutex_), so as far as possible I don't want to break this convention.

[jai1]

Agreed, but the problem is that in PartitionedConsumerImpl:receive we should have released the lock after we checked the state_ and messageListener_ condition since the other data structures are thread safe (see ConsumerImpl::receiveHelper to understand my point)

If you fix this locking then you won't require pendingReceiveMutex_

I am stressing about this minor thing because in all classes we have only one lock to protect all members (mutex_), so as far as possible I don't want to break this convention.

[jai1]

This unlock not required

[jai1]

Why not :

    void PartitionedConsumerImpl::messageReceived(Consumer consumer, const Message& msg) {
        LOG_DEBUG("Received Message from one of the partition - " << msg.impl_->messageId.partition_);
        Lock lock(mutex_);
        if(!pendingReceives_.empty()) {
            ReceiveCallback callback = pendingReceives_.front();
            pendingReceives_.pop();
            lock.unlock();
            callback(ResultOk, msg);
            unAckedMessageTrackerPtr_->add(msg.getMessageId());
            return;
        }
        messages_.push(msg);
        if (messageListener_) {
            listenerExecutor_->postWork(boost::bind(&PartitionedConsumerImpl::internalListener, shared_from_this(), consumer));
        }
    }

[rdhabalia]

for receiveAsync, we need to access pendingReceives_ and messages_ atomically. so, if we release lock before pushing to messages_ then it will face race condition while receiving msg using receiveAsync and callback will never be called.
eg:

• thread-1 comes to receiveAsync() which first checks messages_ if it's empty then it adds callback into pendingReceives_
• thread-2: in this method, if it adds the message into the messages_ same time then : message will sit into messages_ queue and will never be notified for receiveAsync-Callback.

Therefore, we need separate mutex to access pendingReceives_ and messages_ atomically.

[jai1]

No need for notifyPendingReceivedCallback function, just call the callback

[asfgit]

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/pulsar-pull-request/23/

[sijie]

@rdhabalia what is the current state of this PR? are we targeting this for 2.0? If not, can we move it to 2.1?

[sijie]

it seems that this PR was from 1.19.0-incubating. so I will move it out of 2.0, if we need this for 2.0, let's add it back.

[rdhabalia]

sure. I will rebase it and we can take it to next release.