Commit 45a2c9ad authored by Dominik Charousset's avatar Dominik Charousset

Enable blocking request/resp. in test_coordinator

parent 898e5ed0
...@@ -119,12 +119,25 @@ public: ...@@ -119,12 +119,25 @@ public:
/// events (first) and dispatched delayed messages (second). /// events (first) and dispatched delayed messages (second).
std::pair<size_t, size_t> run_dispatch_loop(); std::pair<size_t, size_t> run_dispatch_loop();
/// Executes the next `num` enqueued jobs immediately.
inline void inline_next_enqueues(size_t num) {
inline_enqueues_ += num;
}
/// Executes the next enqueued job immediately.
inline void inline_next_enqueue() {
inline_next_enqueues(1);
}
protected: protected:
void start() override; void start() override;
void stop() override; void stop() override;
void enqueue(resumable* ptr) override; void enqueue(resumable* ptr) override;
private:
size_t inline_enqueues_;
}; };
} // namespace scheduler } // namespace scheduler
......
...@@ -92,7 +92,9 @@ private: ...@@ -92,7 +92,9 @@ private:
} // namespace <anonymous> } // namespace <anonymous>
test_coordinator::test_coordinator(actor_system& sys) : super(sys) { test_coordinator::test_coordinator(actor_system& sys)
: super(sys),
inline_enqueues_(0) {
// nop // nop
} }
void test_coordinator::start() { void test_coordinator::start() {
...@@ -110,7 +112,23 @@ void test_coordinator::stop() { ...@@ -110,7 +112,23 @@ void test_coordinator::stop() {
} }
void test_coordinator::enqueue(resumable* ptr) { void test_coordinator::enqueue(resumable* ptr) {
jobs.push_back(ptr); if (inline_enqueues_ > 0) {
--inline_enqueues_;
dummy_worker worker{this};
switch (ptr->resume(&worker, 1)) {
case resumable::resume_later:
enqueue(ptr);
break;
case resumable::done:
case resumable::awaiting_message:
intrusive_ptr_release(ptr);
break;
case resumable::shutdown_execution_unit:
break;
}
} else {
jobs.push_back(ptr);
}
} }
bool test_coordinator::run_once() { bool test_coordinator::run_once() {
......
...@@ -22,7 +22,7 @@ ...@@ -22,7 +22,7 @@
#include "caf/config.hpp" #include "caf/config.hpp"
#define CAF_SUITE request_response #define CAF_SUITE request_response
#include "caf/test/unit_test.hpp" #include "caf/test/dsl.hpp"
#include "caf/all.hpp" #include "caf/all.hpp"
...@@ -248,7 +248,7 @@ struct fixture { ...@@ -248,7 +248,7 @@ struct fixture {
} // namespace <anonymous> } // namespace <anonymous>
CAF_TEST_FIXTURE_SCOPE(atom_tests, fixture) CAF_TEST_FIXTURE_SCOPE(request_response_tests1, fixture)
CAF_TEST(test_void_res) { CAF_TEST(test_void_res) {
using testee_a = typed_actor<replies_to<int, int>::with<void>>; using testee_a = typed_actor<replies_to<int, int>::with<void>>;
...@@ -481,3 +481,23 @@ CAF_TEST(skip_responses) { ...@@ -481,3 +481,23 @@ CAF_TEST(skip_responses) {
} }
CAF_TEST_FIXTURE_SCOPE_END() CAF_TEST_FIXTURE_SCOPE_END()
CAF_TEST_FIXTURE_SCOPE(request_response_tests2, test_coordinator_fixture<>)
CAF_TEST(request_response_in_test_coordinator) {
auto mirror = sys.spawn<sync_mirror>();
sched.run();
sched.inline_next_enqueue();
// this block would deadlock without inlining the next enqueue
self->request(mirror, infinite, 23).receive(
[](int x) {
CAF_CHECK_EQUAL(x, 23);
},
[&](const error& err) {
CAF_FAIL("unexpected error: " << sys.render(err));
}
);
}
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