[C++] Fix race condition in BlockingQueue

[erobot]

Motivation

BlockingQueue has race condition that can cause threads waiting forever in multithreading environment. ProducerImpl uses BlockingQueue as pendingMessagesQueue_ and can be blocked forever at it. This PR fixes race condition in BlockingQueue.

Race condition details

pulsar/pulsar-client-cpp/lib/BlockingQueue.h

Lines 172 to 185 in 91e2f83

	void pop() {
	Lock lock(mutex_);
	// If the queue is empty, wait until an element is available to be popped
	queueEmptyCondition.wait(lock, QueueNotEmpty<BlockingQueue<T> >(*this));
	bool wasFull = isFullNoMutex();
	queue_.pop_front();
	lock.unlock();
	if (wasFull) {
	// Notify that an element is popped
	queueFullCondition.notify_one();
	}
	}

Use BlockingQueue::Pop as example, its procedure is:

1. lock
2. check wasFull and then change queue state
3. unlock
4. if wasFull, notify one thread waiting at queueFullCondition

Race condition sequence:

1. queue is full and there are multiple threads waiting on queueFullCondition
2. thread A call Pop, lock, wasFull is true, unlock -> queue has one free space
3. thread B call Pop, lock, wasFull is false, unlock -> queue has two free spaces
4. thread A notify one thread waiting at queueFullCondition
5. queue is no loger full again
6. result: except one thread is notified by A, other threads waiting on queueFullCondition are waiting forever

Modifications

• Use notify_all instead of notify_one to notify threads waiting on condition variables
  Reason: Currently only notify threads when queue is full or empty. After unlock, other threads may change queue state, so thread to notify condition can not determine how queue state changed and should use notify_all in case of more then one change occured.

Verifying this change

• Add a test case BlockingQueueTest.testPushPopRace to test concurrent push and pop

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): (no)
• The public API: (no)
• The schema: (no)
• The default values of configurations: (no)
• The wire protocol: (no)
• The rest endpoints: (no)
• The admin cli options: (no)
• Anything that affects deployment: (no)

Documentation

• Does this pull request introduce a new feature? (no)

[merlimat]

Great find!

[zymap]

We have a fix of the integration tests. Please rebase your code. Thanks.

[erobot]

We have a fix of the integration tests. Please rebase your code. Thanks.

Rebase and force push without code changed.

[BewareMyPower]

You need to rebase to the latest master again because a spotbugs check issue was fixed recently by #8819

[erobot]

You need to rebase to the latest master again because a spotbugs check issue was fixed recently by #8819

Rebased again.