[Posix Threads] Refactored condition variables from static to per-instance members and replaced deadlock-prone predicate-based WaitMultiple with polling using try_lock.
This commit is contained in:
@@ -222,64 +222,111 @@ class PosixConditionBase {
|
|||||||
std::chrono::milliseconds timeout) {
|
std::chrono::milliseconds timeout) {
|
||||||
assert_true(!handles.empty());
|
assert_true(!handles.empty());
|
||||||
|
|
||||||
// Construct a condition for all or any depending on wait_all
|
// For single handle, just use the normal Wait path
|
||||||
std::function<bool()> predicate;
|
if (handles.size() == 1) {
|
||||||
{
|
auto result = handles[0]->Wait(timeout);
|
||||||
using iter_t = std::vector<PosixConditionBase*>::const_iterator;
|
return std::make_pair(result, 0);
|
||||||
const auto predicate_inner = [](auto h) { return h->signaled(); };
|
|
||||||
const auto operation =
|
|
||||||
wait_all ? std::all_of<iter_t, decltype(predicate_inner)>
|
|
||||||
: std::any_of<iter_t, decltype(predicate_inner)>;
|
|
||||||
predicate = [&handles, operation, predicate_inner] {
|
|
||||||
return operation(handles.cbegin(), handles.cend(), predicate_inner);
|
|
||||||
};
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// TODO(bwrsandman, Triang3l) This is controversial, see issue #1677
|
// For multiple handles, we need to poll since we can't wait on multiple
|
||||||
// This will probably cause a deadlock on the next thread doing any waiting
|
// condition variables simultaneously. This is a limitation of the POSIX
|
||||||
// if the thread is suspended between locking and waiting
|
// condition variable API.
|
||||||
std::unique_lock lock(mutex_);
|
auto start_time = std::chrono::steady_clock::now();
|
||||||
|
auto end_time = (timeout == std::chrono::milliseconds::max())
|
||||||
|
? std::chrono::steady_clock::time_point::max()
|
||||||
|
: start_time + timeout;
|
||||||
|
|
||||||
bool wait_success = true;
|
while (true) {
|
||||||
// If the timeout is infinite, wait without timeout.
|
// Check all handles to see if any/all are signaled
|
||||||
// The predicate will be checked before beginning the wait
|
// Use try_lock to avoid deadlocks from lock ordering issues
|
||||||
if (timeout == std::chrono::milliseconds::max()) {
|
size_t first_signaled = std::numeric_limits<size_t>::max();
|
||||||
cond_.wait(lock, predicate);
|
bool condition_met = false;
|
||||||
} else {
|
|
||||||
// Wait with timeout.
|
// Try to acquire all locks without blocking
|
||||||
wait_success = cond_.wait_for(lock, timeout, predicate);
|
std::vector<std::unique_lock<std::mutex>> locks;
|
||||||
}
|
locks.reserve(handles.size());
|
||||||
if (wait_success) {
|
bool all_locked = true;
|
||||||
auto first_signaled = std::numeric_limits<size_t>::max();
|
|
||||||
for (auto i = 0u; i < handles.size(); ++i) {
|
for (size_t i = 0; i < handles.size(); ++i) {
|
||||||
if (handles[i]->signaled()) {
|
locks.emplace_back(handles[i]->mutex_, std::try_to_lock);
|
||||||
if (first_signaled > i) {
|
if (!locks.back().owns_lock()) {
|
||||||
first_signaled = i;
|
all_locked = false;
|
||||||
}
|
break;
|
||||||
handles[i]->post_execution();
|
|
||||||
if (!wait_all) break;
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
assert_true(std::numeric_limits<size_t>::max() != first_signaled);
|
|
||||||
return std::make_pair(WaitResult::kSuccess, first_signaled);
|
// If we couldn't acquire all locks, release what we have and retry
|
||||||
|
if (!all_locked) {
|
||||||
|
locks.clear();
|
||||||
|
std::this_thread::yield();
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
|
||||||
|
// Now we have all locks, check the condition
|
||||||
|
if (wait_all) {
|
||||||
|
// For wait_all, check if ALL are signaled
|
||||||
|
bool all_signaled = true;
|
||||||
|
for (size_t i = 0; i < handles.size(); ++i) {
|
||||||
|
if (!handles[i]->signaled()) {
|
||||||
|
all_signaled = false;
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
if (first_signaled == std::numeric_limits<size_t>::max()) {
|
||||||
|
first_signaled = i;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
condition_met = all_signaled;
|
||||||
|
} else {
|
||||||
|
// For wait_any, check if ANY is signaled
|
||||||
|
for (size_t i = 0; i < handles.size(); ++i) {
|
||||||
|
if (handles[i]->signaled()) {
|
||||||
|
first_signaled = i;
|
||||||
|
condition_met = true;
|
||||||
|
break;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
if (condition_met) {
|
||||||
|
// Execute post_execution for the signaled handle(s)
|
||||||
|
if (wait_all) {
|
||||||
|
for (size_t i = 0; i < handles.size(); ++i) {
|
||||||
|
handles[i]->post_execution();
|
||||||
|
}
|
||||||
|
} else {
|
||||||
|
handles[first_signaled]->post_execution();
|
||||||
|
}
|
||||||
|
return std::make_pair(WaitResult::kSuccess, first_signaled);
|
||||||
|
}
|
||||||
|
|
||||||
|
// Release locks before sleeping
|
||||||
|
locks.clear();
|
||||||
|
|
||||||
|
// Check timeout
|
||||||
|
auto now = std::chrono::steady_clock::now();
|
||||||
|
if (now >= end_time) {
|
||||||
|
return std::make_pair<WaitResult, size_t>(WaitResult::kTimeout, 0);
|
||||||
|
}
|
||||||
|
|
||||||
|
// Sleep for a short time before polling again
|
||||||
|
auto remaining =
|
||||||
|
std::chrono::duration_cast<std::chrono::milliseconds>(end_time - now);
|
||||||
|
auto sleep_time = std::min(remaining, std::chrono::milliseconds(1));
|
||||||
|
std::this_thread::sleep_for(sleep_time);
|
||||||
}
|
}
|
||||||
return std::make_pair<WaitResult, size_t>(WaitResult::kTimeout, 0);
|
|
||||||
}
|
}
|
||||||
|
|
||||||
[[nodiscard]] virtual void* native_handle() const {
|
[[nodiscard]] virtual void* native_handle() const {
|
||||||
return cond_.native_handle();
|
return const_cast<std::condition_variable&>(cond_).native_handle();
|
||||||
}
|
}
|
||||||
|
|
||||||
protected:
|
protected:
|
||||||
[[nodiscard]] inline virtual bool signaled() const = 0;
|
[[nodiscard]] inline virtual bool signaled() const = 0;
|
||||||
inline virtual void post_execution() = 0;
|
inline virtual void post_execution() = 0;
|
||||||
static std::condition_variable cond_;
|
std::condition_variable cond_;
|
||||||
static std::mutex mutex_;
|
std::mutex mutex_;
|
||||||
};
|
};
|
||||||
|
|
||||||
std::condition_variable PosixConditionBase::cond_;
|
|
||||||
std::mutex PosixConditionBase::mutex_;
|
|
||||||
|
|
||||||
// There really is no native POSIX handle for a single wait/signal construct
|
// There really is no native POSIX handle for a single wait/signal construct
|
||||||
// pthreads is at a lower level with more handles for such a mechanism.
|
// pthreads is at a lower level with more handles for such a mechanism.
|
||||||
// This simple wrapper class functions as our handle and uses conditional
|
// This simple wrapper class functions as our handle and uses conditional
|
||||||
@@ -373,7 +420,7 @@ class PosixCondition<Mutant> final : public PosixConditionBase {
|
|||||||
}
|
}
|
||||||
|
|
||||||
[[nodiscard]] void* native_handle() const override {
|
[[nodiscard]] void* native_handle() const override {
|
||||||
return mutex_.native_handle();
|
return const_cast<std::mutex&>(mutex_).native_handle();
|
||||||
}
|
}
|
||||||
|
|
||||||
private:
|
private:
|
||||||
@@ -1152,10 +1199,12 @@ void* PosixCondition<Thread>::ThreadStartRoutine(void* parameter) {
|
|||||||
thread->handle_.state_ = State::kFinished;
|
thread->handle_.state_ = State::kFinished;
|
||||||
}
|
}
|
||||||
|
|
||||||
std::unique_lock lock(mutex_);
|
{
|
||||||
thread->handle_.exit_code_ = 0;
|
std::unique_lock lock(thread->handle_.mutex_);
|
||||||
thread->handle_.signaled_ = true;
|
thread->handle_.exit_code_ = 0;
|
||||||
cond_.notify_all();
|
thread->handle_.signaled_ = true;
|
||||||
|
thread->handle_.cond_.notify_all();
|
||||||
|
}
|
||||||
|
|
||||||
current_thread_ = nullptr;
|
current_thread_ = nullptr;
|
||||||
return nullptr;
|
return nullptr;
|
||||||
|
|||||||
Reference in New Issue
Block a user