Commit e9bc9c1f authored by Dominik Charousset's avatar Dominik Charousset

Fix return value of single_reader_queue::try_block

parent 85421ab3
...@@ -91,9 +91,8 @@ public: ...@@ -91,9 +91,8 @@ public:
/// call to {@link try_pop} would succeeed. /// call to {@link try_pop} would succeeed.
/// @pre !closed() /// @pre !closed()
bool can_fetch_more() { bool can_fetch_more() {
if (head_ != nullptr) { if (head_ != nullptr)
return true; return true;
}
auto ptr = stack_.load(); auto ptr = stack_.load();
CAF_ASSERT(ptr != nullptr); CAF_ASSERT(ptr != nullptr);
return ! is_dummy(ptr); return ! is_dummy(ptr);
...@@ -120,10 +119,8 @@ public: ...@@ -120,10 +119,8 @@ public:
/// Tries to set this queue from state `empty` to state `blocked`. /// Tries to set this queue from state `empty` to state `blocked`.
bool try_block() { bool try_block() {
auto e = stack_empty_dummy(); auto e = stack_empty_dummy();
bool res = stack_.compare_exchange_strong(e, reader_blocked_dummy()); return stack_.compare_exchange_strong(e, reader_blocked_dummy());
CAF_ASSERT(e != nullptr); //return res || e == reader_blocked_dummy();
// return true in case queue was already blocked
return res || e == reader_blocked_dummy();
} }
/// Tries to set this queue from state `blocked` to state `empty`. /// Tries to set this queue from state `blocked` to state `empty`.
...@@ -145,9 +142,8 @@ public: ...@@ -145,9 +142,8 @@ public:
template <class F> template <class F>
void close(const F& f) { void close(const F& f) {
clear_cached_elements(f); clear_cached_elements(f);
if (! blocked() && fetch_new_data(nullptr)) { if (! blocked() && fetch_new_data(nullptr))
clear_cached_elements(f); clear_cached_elements(f);
}
cache_.clear(f); cache_.clear(f);
} }
...@@ -156,16 +152,14 @@ public: ...@@ -156,16 +152,14 @@ public:
} }
~single_reader_queue() { ~single_reader_queue() {
if (! closed()) { if (! closed())
close(); close();
}
} }
size_t count(size_t max_count = std::numeric_limits<size_t>::max()) { size_t count(size_t max_count = std::numeric_limits<size_t>::max()) {
size_t res = cache_.count(max_count); size_t res = cache_.count(max_count);
if (res >= max_count) { if (res >= max_count)
return res; return res;
}
fetch_new_data(); fetch_new_data();
auto ptr = head_; auto ptr = head_;
while (ptr && res < max_count) { while (ptr && res < max_count) {
...@@ -212,9 +206,8 @@ public: ...@@ -212,9 +206,8 @@ public:
CAF_ASSERT(! closed()); CAF_ASSERT(! closed());
if (! can_fetch_more() && try_block()) { if (! can_fetch_more() && try_block()) {
std::unique_lock<Mutex> guard(mtx); std::unique_lock<Mutex> guard(mtx);
while (blocked()) { while (blocked())
cv.wait(guard); cv.wait(guard);
}
} }
} }
......
...@@ -691,9 +691,8 @@ void local_actor::launch(execution_unit* eu, bool lazy, bool hide) { ...@@ -691,9 +691,8 @@ void local_actor::launch(execution_unit* eu, bool lazy, bool hide) {
CAF_ASSERT(eu != nullptr); CAF_ASSERT(eu != nullptr);
// do not schedule immediately when spawned with `lazy_init` // do not schedule immediately when spawned with `lazy_init`
// mailbox could be set to blocked // mailbox could be set to blocked
if (lazy && mailbox().try_block()) { if (lazy && mailbox().try_block())
return; return;
}
// scheduler has a reference count to the actor as long as // scheduler has a reference count to the actor as long as
// it is waiting to get scheduled // it is waiting to get scheduled
intrusive_ptr_add_ref(ctrl()); intrusive_ptr_add_ref(ctrl());
...@@ -782,9 +781,8 @@ resumable::resume_result local_actor::resume(execution_unit* eu, ...@@ -782,9 +781,8 @@ resumable::resume_result local_actor::resume(execution_unit* eu,
} else { } else {
CAF_LOG_DEBUG("no more element in mailbox; going to block"); CAF_LOG_DEBUG("no more element in mailbox; going to block");
reset_timeout_if_needed(); reset_timeout_if_needed();
if (mailbox().try_block()) { if (mailbox().try_block())
return resumable::awaiting_message; return resumable::awaiting_message;
}
CAF_LOG_DEBUG("try_block() interrupted by new message"); CAF_LOG_DEBUG("try_block() interrupted by new message");
} }
} }
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment