From a734f77757287abb479ac180e64b25a6182d8fdb Mon Sep 17 00:00:00 2001 From: Lingxuan Zuo Date: Fri, 19 Jun 2020 12:16:40 +0800 Subject: [PATCH] [Streaming] fix cv hang in multithread variables race (#8984) --- streaming/src/event_service.cc | 37 ++++++++++++++--------- streaming/src/event_service.h | 14 ++++++--- streaming/src/test/event_service_tests.cc | 17 +++++------ 3 files changed, 40 insertions(+), 28 deletions(-) diff --git a/streaming/src/event_service.cc b/streaming/src/event_service.cc index 2e788c9fe..a6dd75744 100644 --- a/streaming/src/event_service.cc +++ b/streaming/src/event_service.cc @@ -1,3 +1,4 @@ +#include #include #include "event_service.h" @@ -6,22 +7,22 @@ namespace ray { namespace streaming { EventQueue::~EventQueue() { - is_freezed_ = false; + is_active_ = false; no_full_cv_.notify_all(); no_empty_cv_.notify_all(); }; -void EventQueue::Unfreeze() { is_freezed_ = true; } +void EventQueue::Unfreeze() { is_active_ = true; } void EventQueue::Freeze() { - is_freezed_ = false; + is_active_ = false; no_empty_cv_.notify_all(); no_full_cv_.notify_all(); } void EventQueue::Push(const Event &t) { std::unique_lock lock(ring_buffer_mutex_); - while (Size() >= capacity_ && is_freezed_) { + while (Size() >= capacity_ && is_active_) { STREAMING_LOG(WARNING) << " EventQueue is full, its size:" << Size() << " capacity:" << capacity_ << " buffer size:" << buffer_.size() @@ -29,7 +30,7 @@ void EventQueue::Push(const Event &t) { no_full_cv_.wait(lock); STREAMING_LOG(WARNING) << "Event server is full_sleep be notified"; } - if (!is_freezed_) { + if (!is_active_) { return; } if (t.urgent) { @@ -56,12 +57,22 @@ void EventQueue::Pop() { no_full_cv_.notify_all(); } +void EventQueue::WaitFor(std::unique_lock &lock) { + // To avoid deadlock when EventQueue is empty but is_active is changed in other + // thread, Event queue should awaken this condtion variable and check it again. + while (is_active_ && Empty()) { + if (!no_empty_cv_.wait_for(lock, std::chrono::milliseconds(kConditionTimeoutMs), + [this]() { return !is_active_ || !Empty(); })) { + STREAMING_LOG(DEBUG) << "No empty condition variable wait timeout." + << " Empty => " << Empty() << ", is active " << is_active_; + } + } +} + bool EventQueue::Get(Event &evt) { std::unique_lock lock(ring_buffer_mutex_); - while (Empty() && is_freezed_) { - no_empty_cv_.wait(lock); - } - if (!is_freezed_) { + WaitFor(lock); + if (!is_active_) { return false; } if (!urgent_buffer_.empty()) { @@ -76,11 +87,9 @@ bool EventQueue::Get(Event &evt) { Event EventQueue::PopAndGet() { std::unique_lock lock(ring_buffer_mutex_); - while (Empty() && is_freezed_) { - no_empty_cv_.wait(lock); - } - if (!is_freezed_) { - // Return error event if queue is freezed. + WaitFor(lock); + if (!is_active_) { + // Return error event if queue is active. return Event({nullptr, EventType::ErrorEvent, false}); } if (!urgent_buffer_.empty()) { diff --git a/streaming/src/event_service.h b/streaming/src/event_service.h index 03a08385a..298443800 100644 --- a/streaming/src/event_service.h +++ b/streaming/src/event_service.h @@ -24,7 +24,7 @@ enum class EventType : uint8_t { FullChannel = 3, // Recovery at the beginning. Reload = 4, - // Error event if event queue is freezed. + // Error event if event queue is not active. ErrorEvent = 5 }; @@ -50,14 +50,14 @@ struct Event { /// processing functions ordered by its priority. class EventQueue { public: - EventQueue(size_t size) : urgent_(false), capacity_(size), is_freezed_(true) {} + EventQueue(size_t size) : urgent_(false), capacity_(size), is_active_(true) {} virtual ~EventQueue(); /// Resume event queue to normal model. void Unfreeze(); - /// Push is prohibited when event queue is freezed. + /// Push is prohibited when event queue is not active. void Freeze(); void Push(const Event &t); @@ -84,6 +84,9 @@ class EventQueue { inline bool Full() const { return buffer_.size() + urgent_buffer_.size() == capacity_; } + /// Wait for queue util it's timeout or any stuff in. + void WaitFor(std::unique_lock &lock); + private: std::mutex ring_buffer_mutex_; std::condition_variable no_empty_cv_; @@ -95,7 +98,10 @@ class EventQueue { // Urgent event will be poped out first if urgent_ flag is true. bool urgent_; size_t capacity_; - bool is_freezed_; + // Event service active flag. + bool is_active_; + // Pop/Get timeout ms for condition variables wait. + static constexpr int kConditionTimeoutMs = 200; }; class EventService { diff --git a/streaming/src/test/event_service_tests.cc b/streaming/src/test/event_service_tests.cc index 119e1d4d3..a7f4b1233 100644 --- a/streaming/src/test/event_service_tests.cc +++ b/streaming/src/test/event_service_tests.cc @@ -10,7 +10,6 @@ bool SendEmptyToChannel(ProducerChannelInfo *info) { return true; } /// Mock function for write all messages to channel. bool WriteAllToChannel(ProducerChannelInfo *info) { return true; } -bool stop = false; TEST(EventServiceTest, Test1) { std::shared_ptr server = std::make_shared(); @@ -19,30 +18,28 @@ TEST(EventServiceTest, Test1) { server->Register(EventType::UserEvent, WriteAllToChannel); server->Register(EventType::FlowEvent, WriteAllToChannel); - std::thread thread_empty([server, &mock_channel_info] { + bool stop = false; + std::thread thread_empty([server, &mock_channel_info, &stop] { std::chrono::milliseconds MockTimer(20); - while (true) { - if (stop) break; + while (!stop) { Event event{&mock_channel_info, EventType::EmptyEvent, true}; server->Push(event); std::this_thread::sleep_for(MockTimer); } }); - std::thread thread_flow([server, &mock_channel_info] { + std::thread thread_flow([server, &mock_channel_info, &stop] { std::chrono::milliseconds MockTimer(2); - while (true) { - if (stop) break; + while (!stop) { Event event{&mock_channel_info, EventType::FlowEvent, true}; server->Push(event); std::this_thread::sleep_for(MockTimer); } }); - std::thread thread_user([server, &mock_channel_info] { + std::thread thread_user([server, &mock_channel_info, &stop] { std::chrono::milliseconds MockTimer(2); - while (true) { - if (stop) break; + while (!stop) { Event event{&mock_channel_info, EventType::UserEvent, true}; server->Push(event); std::this_thread::sleep_for(MockTimer);