Unverified Commit acf91eb2 authored by Noir's avatar Noir Committed by GitHub

Merge pull request #1244

Properly handle request.await in test DSL
parents 83144dce 19db4a86
...@@ -34,6 +34,9 @@ is based on [Keep a Changelog](https://keepachangelog.com). ...@@ -34,6 +34,9 @@ is based on [Keep a Changelog](https://keepachangelog.com).
- Fix heap-use-after-free when accessing the meta objects table in applications - Fix heap-use-after-free when accessing the meta objects table in applications
that leave the `main` function while the actor system and its worker threads that leave the `main` function while the actor system and its worker threads
are still running (#1241). are still running (#1241).
- The testing DSL now properly accounts for the message prioritization of actors
(suspending regular behavior until receiving the response) when using
`request.await` (#1232).
## [0.18.1] - 2021-03-19 ## [0.18.1] - 2021-03-19
......
...@@ -102,6 +102,13 @@ public: ...@@ -102,6 +102,13 @@ public:
list_.peek_all(f); list_.peek_all(f);
} }
/// Tries to find the next element in the queue (excluding cached elements)
/// that matches the given predicate.
template <class Predicate>
pointer find_if(Predicate pred) {
return list_.find_if(pred);
}
// -- modifiers ------------------------------------------------------------- // -- modifiers -------------------------------------------------------------
/// Removes all elements from the queue. /// Removes all elements from the queue.
......
...@@ -161,6 +161,13 @@ public: ...@@ -161,6 +161,13 @@ public:
return queue_.peek(); return queue_.peek();
} }
/// Tries to find an element in the queue that matches the given predicate.
template <class Predicate>
pointer find_if(Predicate pred) {
fetch_more();
return queue_.find_if(pred);
}
queue_type& queue() noexcept { queue_type& queue() noexcept {
return queue_; return queue_;
} }
......
...@@ -124,6 +124,15 @@ public: ...@@ -124,6 +124,15 @@ public:
f(*i); f(*i);
} }
/// Tries to find an element in the queue that matches the given predicate.
template <class Predicate>
pointer find_if(Predicate pred) {
for (auto i = begin(); i != end(); ++i)
if (pred(*i))
return promote(i.ptr);
return nullptr;
}
// -- modifiers ------------------------------------------------------------- // -- modifiers -------------------------------------------------------------
/// Removes all elements from the queue. /// Removes all elements from the queue.
......
...@@ -154,6 +154,15 @@ public: ...@@ -154,6 +154,15 @@ public:
kvp.second.peek_all(f); kvp.second.peek_all(f);
} }
/// Tries to find an element in the queue that matches the given predicate.
template <class Predicate>
pointer find_if(Predicate pred) {
for (auto& kvp : qs_)
if (auto ptr = kvp.second.find_if(pred))
return ptr;
return nullptr;
}
/// Returns `true` if all queues are empty, `false` otherwise. /// Returns `true` if all queues are empty, `false` otherwise.
bool empty() const noexcept { bool empty() const noexcept {
return total_task_size() == 0; return total_task_size() == 0;
......
...@@ -80,6 +80,12 @@ public: ...@@ -80,6 +80,12 @@ public:
return peek_recursion<0>(); return peek_recursion<0>();
} }
/// Tries to find an element in the queue that matches the given predicate.
template <class Predicate>
pointer find_if(Predicate pred) {
return find_if_recursion<0>(pred);
}
/// Applies `f` to each element in the queue. /// Applies `f` to each element in the queue.
template <class F> template <class F>
void peek_all(F f) const { void peek_all(F f) const {
...@@ -204,6 +210,20 @@ private: ...@@ -204,6 +210,20 @@ private:
return peek_recursion<I + 1>(); return peek_recursion<I + 1>();
} }
template <size_t I, class Predicate>
detail::enable_if_t<I == num_queues, pointer> find_if_recursion(Predicate) {
return nullptr;
}
template <size_t I, class Predicate>
detail::enable_if_t<I != num_queues, pointer>
find_if_recursion(Predicate pred) {
if (auto ptr = std::get<I>(qs_).find_if(pred))
return ptr;
else
return find_if_recursion<I + 1>(std::move(pred));
}
template <size_t I, class F> template <size_t I, class F>
detail::enable_if_t<I == num_queues> peek_all_recursion(F&) const { detail::enable_if_t<I == num_queues> peek_all_recursion(F&) const {
// nop // nop
......
...@@ -175,8 +175,17 @@ void scheduled_actor::enqueue(mailbox_element_ptr ptr, execution_unit* eu) { ...@@ -175,8 +175,17 @@ void scheduled_actor::enqueue(mailbox_element_ptr ptr, execution_unit* eu) {
break; break;
} }
} }
mailbox_element* scheduled_actor::peek_at_next_mailbox_element() { mailbox_element* scheduled_actor::peek_at_next_mailbox_element() {
return mailbox().closed() || mailbox().blocked() ? nullptr : mailbox().peek(); if (mailbox().closed() || mailbox().blocked()) {
return nullptr;
} else if (awaited_responses_.empty()) {
return mailbox().peek();
} else {
auto mid = awaited_responses_.begin()->first;
auto pred = [mid](mailbox_element& x) { return x.mid == mid; };
return mailbox().find_if(pred);
}
} }
// -- overridden functions of local_actor -------------------------------------- // -- overridden functions of local_actor --------------------------------------
......
...@@ -6,7 +6,7 @@ ...@@ -6,7 +6,7 @@
#include "caf/mixin/requester.hpp" #include "caf/mixin/requester.hpp"
#include "caf/test/dsl.hpp" #include "core-test.hpp"
#include <numeric> #include <numeric>
...@@ -187,4 +187,37 @@ CAF_TEST(exceptions while processing requests trigger error messages) { ...@@ -187,4 +187,37 @@ CAF_TEST(exceptions while processing requests trigger error messages) {
#endif // CAF_ENABLE_EXCEPTIONS #endif // CAF_ENABLE_EXCEPTIONS
SCENARIO("request.await enforces a processing order") {
GIVEN("an actor that is waiting for a request.await handler") {
auto server = sys.spawn([]() -> behavior {
return {
[](int32_t x) { return x * x; },
};
});
run();
auto client = sys.spawn([server](event_based_actor* self) -> behavior {
self->request(server, infinite, int32_t{3}).await([](int32_t res) {
CHECK_EQ(res, 9);
});
return {
[](const std::string& str) {
// Received from self.
CHECK_EQ(str, "hello");
},
};
});
sched.run_once();
WHEN("sending it a message before the response arrives") {
THEN("the actor handles the asynchronous message later") {
self->send(client, "hello");
disallow((std::string), from(self).to(client)); // not processed yet
expect((int32_t), from(client).to(server).with(3)); // client -> server
disallow((std::string), from(self).to(client)); // not processed yet
expect((int32_t), from(server).to(client).with(9)); // server -> client
expect((std::string), from(self).to(client).with("hello")); // at last
}
}
}
}
CAF_TEST_FIXTURE_SCOPE_END() CAF_TEST_FIXTURE_SCOPE_END()
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