Commit ee447203 authored by Dominik Charousset's avatar Dominik Charousset

Fix issue in http_broker test during shutdown

parent 4a97de85
...@@ -122,6 +122,9 @@ public: ...@@ -122,6 +122,9 @@ public:
/// Returns `true` if a `runnable` was available, `false` otherwise. /// Returns `true` if a `runnable` was available, `false` otherwise.
bool try_exec_runnable(); bool try_exec_runnable();
/// Executes all pending `runnable` objects.
void flush_runnables();
protected: protected:
void dispatch_runnable(runnable_ptr ptr) override; void dispatch_runnable(runnable_ptr ptr) override;
......
...@@ -268,11 +268,11 @@ void test_multiplexer::read_data(connection_handle hdl) { ...@@ -268,11 +268,11 @@ void test_multiplexer::read_data(connection_handle hdl) {
break; break;
case receive_policy_flag::at_most: case receive_policy_flag::at_most:
auto max_bytes = static_cast<ptrdiff_t>(sd.recv_conf.second); auto max_bytes = static_cast<ptrdiff_t>(sd.recv_conf.second);
while (sd.xbuf.size() > 0) { while (! sd.xbuf.empty()) {
sd.rd_buf.clear(); sd.rd_buf.clear();
auto xbuf_size = static_cast<ptrdiff_t>(sd.xbuf.size());
auto first = sd.xbuf.begin(); auto first = sd.xbuf.begin();
auto last = (max_bytes < sd.xbuf.size()) ? first + max_bytes auto last = (max_bytes < xbuf_size) ? first + max_bytes : sd.xbuf.end();
: sd.xbuf.end();
sd.rd_buf.insert(sd.rd_buf.end(), first, last); sd.rd_buf.insert(sd.rd_buf.end(), first, last);
sd.xbuf.erase(first, last); sd.xbuf.erase(first, last);
sd.ptr->consume(sd.rd_buf.data(), sd.rd_buf.size()); sd.ptr->consume(sd.rd_buf.data(), sd.rd_buf.size());
...@@ -312,6 +312,18 @@ bool test_multiplexer::try_exec_runnable() { ...@@ -312,6 +312,18 @@ bool test_multiplexer::try_exec_runnable() {
return true; return true;
} }
void test_multiplexer::flush_runnables() {
std::vector<runnable_ptr> runnables;
runnables.reserve(256);
{ // critical section
guard_type guard{mx_};
while (! runnables_.empty())
runnables.emplace_back(std::move(runnables_.front()));
}
for (auto& ptr : runnables)
ptr->run();
}
void test_multiplexer::dispatch_runnable(runnable_ptr ptr) { void test_multiplexer::dispatch_runnable(runnable_ptr ptr) {
std::list<runnable_ptr> tmp; std::list<runnable_ptr> tmp;
tmp.emplace_back(std::move(ptr)); tmp.emplace_back(std::move(ptr));
......
...@@ -200,9 +200,10 @@ public: ...@@ -200,9 +200,10 @@ public:
~fixture() { ~fixture() {
anon_send_exit(aut_, exit_reason::kill); anon_send_exit(aut_, exit_reason::kill);
// run the exit message explicitly, since we do not invoke any "IO" // run the exit message and other pending messages explicitly,
// from this point on that would trigger the exit message implicitly // since we do not invoke any "IO" from this point on that would
mpx_->exec_runnable(); // trigger the exit message implicitly
mpx_->flush_runnables();
await_all_actors_done(); await_all_actors_done();
shutdown(); shutdown();
} }
...@@ -216,12 +217,13 @@ public: ...@@ -216,12 +217,13 @@ public:
mock_t(const mock_t&) = default; mock_t(const mock_t&) = default;
mock_t& expect(const std::string& what) { mock_t& expect(const std::string& x) {
auto& buf = this_->mpx_->output_buffer(this_->connection_); auto& buf = this_->mpx_->output_buffer(this_->connection_);
CAF_REQUIRE((buf.size() >= what.size())); CAF_REQUIRE((buf.size() >= x.size()));
CAF_REQUIRE((std::equal(buf.begin(), buf.begin() + what.size(), CAF_REQUIRE((std::equal(buf.begin(),
what.begin())); buf.begin() + static_cast<ptrdiff_t>(x.size()),
buf.erase(buf.begin(), buf.begin() + what.size())); x.begin()));
buf.erase(buf.begin(), buf.begin() + static_cast<ptrdiff_t>(x.size())));
return *this; return *this;
} }
......
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