You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "BewareMyPower (via GitHub)" <gi...@apache.org> on 2023/10/26 05:03:17 UTC
[PR] Fix possible deadlock of Future when adding a listener after completed [pulsar-client-cpp]
BewareMyPower opened a new pull request, #334:
URL: https://github.com/apache/pulsar-client-cpp/pull/334
### Motivation
There is a case that deadlock could happen for a `Future`. Assume there is a `Promise` and its `Future`.
1. Call `Future::addListener` to add a listener that tries to acquire a user-provided mutex (`lock`).
2. Thread 1: Acquire `lock` first.
3. Thread 2: Call `Promise::setValue`, the listener will be triggered first before completed. Since `lock` is held by Thread 1, the listener will be blocked.
4. Thread 1: Call `Future::addListener`, since it detects the `InternalState::completed_` is true, it will call `get` to retrieve the result and value.
Then, deadlock happens:
- Thread 1 waits for `lock` is released, and then complete `InternalState::future_`.
- Thread 2 holds `lock` but wait for `InternalState::future_` is completed.
In a real world case, if we acquire a lock before
`ProducerImpl::closeAsync`, then another thread call `setValue` in `ClientConnection::handleSuccess` and the callback of `createProducerAsync` tries to acquire the lock, `handleSuccess` will be blocked. Then in `closeAsync`, the current thread will be blocked in:
```c++
cnx->sendRequestWithId(Commands::newCloseProducer(producerId_, requestId), requestId)
.addListener([self, callback](Result result, const ResponseData&) { callback(result); });
```
The stacks:
```
Thread 1:
#11 0x00007fab80da2173 in pulsar::InternalState<...>::complete (this=0x3d53e7a10, result=..., value=...) at lib/Futre.h:61
#13 pulsar::ClientConnection::handleSuccess (this=this@entry=0x2214bc000, success=...) at lib/ClientConnection.cc:1552
Thread 2:
#8 get (result=..., this=0x3d53e7a10) at lib/Future.h:69
#9 pulsar::InternalState<...>::addListener (this=this@entry=0x3d53e7a10, listener=...) at lib/Future.h:51
#11 0x00007fab80e8dc4e in pulsar::ProducerImpl::closeAsync at lib/ProducerImpl.cc:794
```
There are two points that make the deadlock:
1. We use `completed_` to represent if the future is completed. However, after it's true, the future might not be completed because the value is not set and the listeners are not completed.
2. If `addListener` is called after it's completed, we still push the listener to `listeners_` so that previous listeners could be executed before the new listener. This guarantee is unnecessarily strong.
### Modifications
First, complete the future before calling the listeners.
Then, use an enum to represent the status:
- INITIAL: `complete` has not been called
- COMPLETING: when the 1st time `complete` is called, the status will change from INITIAL to COMPLETING
- COMPLETED: the future is completed.
Besides, implementation of `Future` is simplified. https://github.com/apache/pulsar-client-cpp/pull/299 fixes a possible mutex crash by introducing the `std::future`. However, the root cause is the conditional variable is not used correctly:
> Even if the shared variable is atomic, it must be modified while owning the mutex to correctly publish the modification to the waiting thread.
See https://en.cppreference.com/w/cpp/thread/condition_variable
The simplest way to fix
https://github.com/apache/pulsar-client-cpp/pull/298 is just adding `lock.lock()` before `state->condition.notify_all();`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix possible deadlock of Future when adding a listener after completed [pulsar-client-cpp]
Posted by "shibd (via GitHub)" <gi...@apache.org>.
shibd commented on code in PR #334:
URL: https://github.com/apache/pulsar-client-cpp/pull/334#discussion_r1376254895
##########
lib/Future.h:
##########
@@ -38,71 +35,70 @@ class InternalState {
using Pair = std::pair<Result, Type>;
using Lock = std::unique_lock<std::mutex>;
+ enum Status : uint8_t
+ {
+ INITIAL,
+ COMPLETING,
+ COMPLETED
+ };
+
// NOTE: Add the constructor explicitly just to be compatible with GCC 4.8
InternalState() {}
void addListener(Listener listener) {
Lock lock{mutex_};
- listeners_.emplace_back(listener);
- lock.unlock();
-
if (completed()) {
- Type value;
- Result result = get(value);
- triggerListeners(result, value);
+ auto result = result_;
+ auto value = value_;
+ lock.unlock();
+ listener(result, value);
+ } else {
+ tailListener_ = listeners_.emplace_after(tailListener_, std::move(listener));
}
}
bool complete(Result result, const Type &value) {
- bool expected = false;
- if (!completed_.compare_exchange_strong(expected, true)) {
+ Status expected = Status::INITIAL;
+ if (!status_.compare_exchange_strong(expected, Status::COMPLETING)) {
return false;
}
- triggerListeners(result, value);
- promise_.set_value(std::make_pair(result, value));
- return true;
- }
-
- bool completed() const noexcept { return completed_; }
- Result get(Type &result) {
- const auto &pair = future_.get();
- result = pair.second;
- return pair.first;
- }
+ // Ensure if another thread calls `addListener` at the same time, that thread can get the value by
+ // `get` before the existing listeners are executed
+ Lock lock{mutex_};
+ result_ = result;
+ value_ = value;
+ status_ = COMPLETED;
+ cond_.notify_all();
- // Only public for test
- void triggerListeners(Result result, const Type &value) {
- while (true) {
- Lock lock{mutex_};
- if (listeners_.empty()) {
- return;
+ if (!listeners_.empty()) {
+ auto listeners = std::move(listeners_);
+ lock.unlock();
+ for (auto &&listener : listeners) {
+ listener(result, value);
}
+ }
- bool expected = false;
- if (!listenerRunning_.compare_exchange_strong(expected, true)) {
- // There is another thread that polled a listener that is running, skip polling and release
- // the lock. Here we wait for some time to avoid busy waiting.
- std::this_thread::sleep_for(std::chrono::milliseconds(1));
- continue;
- }
- auto listener = std::move(listeners_.front());
- listeners_.pop_front();
- lock.unlock();
+ return true;
+ }
- listener(result, value);
- listenerRunning_ = false;
- }
+ bool completed() const noexcept { return status_.load() == COMPLETED; }
+
+ Result get(Type &value) const {
+ Lock lock{mutex_};
+ cond_.wait(lock, [this] { return completed(); });
+ value = value_;
+ return result_;
}
private:
- std::atomic_bool completed_{false};
- std::promise<Pair> promise_;
- std::shared_future<Pair> future_{promise_.get_future()};
-
- std::list<Listener> listeners_;
mutable std::mutex mutex_;
- std::atomic_bool listenerRunning_{false};
+ mutable std::condition_variable cond_;
+ std::forward_list<Listener> listeners_;
+ decltype(listeners_.before_begin()) tailListener_{listeners_.before_begin()};
Review Comment:
What is the purpose of `tailListener_ `?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix possible deadlock of Future when adding a listener after completed [pulsar-client-cpp]
Posted by "merlimat (via GitHub)" <gi...@apache.org>.
merlimat merged PR #334:
URL: https://github.com/apache/pulsar-client-cpp/pull/334
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix possible deadlock of Future when adding a listener after completed [pulsar-client-cpp]
Posted by "shibd (via GitHub)" <gi...@apache.org>.
shibd commented on code in PR #334:
URL: https://github.com/apache/pulsar-client-cpp/pull/334#discussion_r1376254895
##########
lib/Future.h:
##########
@@ -38,71 +35,70 @@ class InternalState {
using Pair = std::pair<Result, Type>;
using Lock = std::unique_lock<std::mutex>;
+ enum Status : uint8_t
+ {
+ INITIAL,
+ COMPLETING,
+ COMPLETED
+ };
+
// NOTE: Add the constructor explicitly just to be compatible with GCC 4.8
InternalState() {}
void addListener(Listener listener) {
Lock lock{mutex_};
- listeners_.emplace_back(listener);
- lock.unlock();
-
if (completed()) {
- Type value;
- Result result = get(value);
- triggerListeners(result, value);
+ auto result = result_;
+ auto value = value_;
+ lock.unlock();
+ listener(result, value);
+ } else {
+ tailListener_ = listeners_.emplace_after(tailListener_, std::move(listener));
}
}
bool complete(Result result, const Type &value) {
- bool expected = false;
- if (!completed_.compare_exchange_strong(expected, true)) {
+ Status expected = Status::INITIAL;
+ if (!status_.compare_exchange_strong(expected, Status::COMPLETING)) {
return false;
}
- triggerListeners(result, value);
- promise_.set_value(std::make_pair(result, value));
- return true;
- }
-
- bool completed() const noexcept { return completed_; }
- Result get(Type &result) {
- const auto &pair = future_.get();
- result = pair.second;
- return pair.first;
- }
+ // Ensure if another thread calls `addListener` at the same time, that thread can get the value by
+ // `get` before the existing listeners are executed
+ Lock lock{mutex_};
+ result_ = result;
+ value_ = value;
+ status_ = COMPLETED;
+ cond_.notify_all();
- // Only public for test
- void triggerListeners(Result result, const Type &value) {
- while (true) {
- Lock lock{mutex_};
- if (listeners_.empty()) {
- return;
+ if (!listeners_.empty()) {
+ auto listeners = std::move(listeners_);
+ lock.unlock();
+ for (auto &&listener : listeners) {
+ listener(result, value);
}
+ }
- bool expected = false;
- if (!listenerRunning_.compare_exchange_strong(expected, true)) {
- // There is another thread that polled a listener that is running, skip polling and release
- // the lock. Here we wait for some time to avoid busy waiting.
- std::this_thread::sleep_for(std::chrono::milliseconds(1));
- continue;
- }
- auto listener = std::move(listeners_.front());
- listeners_.pop_front();
- lock.unlock();
+ return true;
+ }
- listener(result, value);
- listenerRunning_ = false;
- }
+ bool completed() const noexcept { return status_.load() == COMPLETED; }
+
+ Result get(Type &value) const {
+ Lock lock{mutex_};
+ cond_.wait(lock, [this] { return completed(); });
+ value = value_;
+ return result_;
}
private:
- std::atomic_bool completed_{false};
- std::promise<Pair> promise_;
- std::shared_future<Pair> future_{promise_.get_future()};
-
- std::list<Listener> listeners_;
mutable std::mutex mutex_;
- std::atomic_bool listenerRunning_{false};
+ mutable std::condition_variable cond_;
+ std::forward_list<Listener> listeners_;
+ decltype(listeners_.before_begin()) tailListener_{listeners_.before_begin()};
Review Comment:
What is the purpose of `tailListener_ `?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix possible deadlock of Future when adding a listener after completed [pulsar-client-cpp]
Posted by "BewareMyPower (via GitHub)" <gi...@apache.org>.
BewareMyPower commented on PR #334:
URL: https://github.com/apache/pulsar-client-cpp/pull/334#issuecomment-1780552628
The `Oauth2test` seems stuck at connecting:
```
2023-10-26 07:13:58.425 DEBUG [140675173545536] AuthOauth2:378 | access_token: *** expires_in: 10
2023-10-26 07:13:58.447 DEBUG [140675173545536] ClientConnection:882 | [[::1]:36014 -> [::1]:6650] Handling incoming command: CONNECTED
2023-10-26 07:13:58.447 DEBUG [140675173545536] ClientConnection:279 | Connection has max message size setting: 5[242](https://github.com/apache/pulsar-client-cpp/actions/runs/6650646879/job/18071147259?pr=334#step:8:243)880
2023-10-26 07:13:58.447 DEBUG [140675173545536] ClientConnection:281 | Current max message size is: 5242880
2023-10-26 07:14:28.118 DEBUG [140675173545536] ClientConnection:882 | [[::1]:36014 -> [::1]:6650] Handling incoming command: PING
2023-10-26 07:14:28.118 DEBUG [140675173545536] ClientConnection:955 | [[::1]:36014 -> [::1]:6650] Replying to ping command
2023-10-26 07:14:28.447 DEBUG [140675173545536] ClientConnection:1232 | [[::1]:36014 -> [::1]:6650] Sending ping message
2023-10-26 07:14:28.450 DEBUG [140675173545536] ClientConnection:882 | [[::1]:36014 -> [::1]:6650] Handling incoming command: PONG
```
However, I cannot reproduce it locally. I'm marking it as drafted and going to do some investigation
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org