Commit 92f040cc authored by Dominik Charousset's avatar Dominik Charousset

Fix leak in stream serv and its test

parent 9b98f73d
...@@ -54,22 +54,22 @@ void streamer_impl(event_based_actor* self, const actor& dest) { ...@@ -54,22 +54,22 @@ void streamer_impl(event_based_actor* self, const actor& dest) {
// destination of the stream // destination of the stream
dest, dest,
// initialize state // initialize state
[&](buf& xs) { [](buf& xs) {
xs = buf{1, 2, 3, 4, 5, 6, 7, 8, 9}; xs = buf{1, 2, 3, 4, 5, 6, 7, 8, 9};
}, },
// get next element // get next element
[=](buf& xs, downstream<int>& out, size_t num) { [](buf& xs, downstream<int>& out, size_t num) {
auto n = std::min(num, xs.size()); auto n = std::min(num, xs.size());
for (size_t i = 0; i < n; ++i) for (size_t i = 0; i < n; ++i)
out.push(xs[i]); out.push(xs[i]);
xs.erase(xs.begin(), xs.begin() + static_cast<ptrdiff_t>(n)); xs.erase(xs.begin(), xs.begin() + static_cast<ptrdiff_t>(n));
}, },
// check whether we reached the end // check whether we reached the end
[=](const buf& xs) { [](const buf& xs) {
return xs.empty(); return xs.empty();
}, },
// handle result of the stream // handle result of the stream
[=](expected<int>) { [](expected<int>) {
// nop // nop
} }
); );
...@@ -146,6 +146,13 @@ public: ...@@ -146,6 +146,13 @@ public:
}; };
} }
void on_exit() override {
CAF_CHECK_EQUAL(incoming_.num_streams(), 0u);
CAF_CHECK_EQUAL(outgoing_.num_streams(), 0u);
CAF_CHECK(streams().empty());
remotes().empty();
}
strong_actor_ptr remote_stream_serv(const node_id& nid) override; strong_actor_ptr remote_stream_serv(const node_id& nid) override;
private: private:
...@@ -157,7 +164,7 @@ private: ...@@ -157,7 +164,7 @@ private:
// Simulates a regular forwarding_actor_proxy by pushing a handle to the // Simulates a regular forwarding_actor_proxy by pushing a handle to the
// original to the forwarding stack and redirecting each message to the // original to the forwarding stack and redirecting each message to the
// stream_serv. // stream_serv.
class pseudo_proxy : public raw_event_based_actor{ class pseudo_proxy : public raw_event_based_actor {
public: public:
pseudo_proxy(actor_config& cfg, actor stream_serv, actor original) pseudo_proxy(actor_config& cfg, actor stream_serv, actor original)
: raw_event_based_actor(cfg), : raw_event_based_actor(cfg),
......
...@@ -255,14 +255,14 @@ strong_actor_ptr middleman::remote_lookup(atom_value name, const node_id& nid) { ...@@ -255,14 +255,14 @@ strong_actor_ptr middleman::remote_lookup(atom_value name, const node_id& nid) {
void middleman::start() { void middleman::start() {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
// create hooks // Create hooks.
for (auto& f : system().config().hook_factories) for (auto& f : system().config().hook_factories)
hooks_.emplace_back(f(system_)); hooks_.emplace_back(f(system_));
// launch backend // Launch backend.
backend_supervisor_ = backend().make_supervisor(); backend_supervisor_ = backend().make_supervisor();
if (!backend_supervisor_) { if (!backend_supervisor_) {
// the only backend that returns a `nullptr` is the `test_multiplexer` // The only backend that returns a `nullptr` is the `test_multiplexer`
// which does not have its own thread but uses the main thread instead // which does not have its own thread but uses the main thread instead.
backend().thread_id(std::this_thread::get_id()); backend().thread_id(std::this_thread::get_id());
} else { } else {
thread_ = std::thread{[this] { thread_ = std::thread{[this] {
...@@ -272,9 +272,7 @@ void middleman::start() { ...@@ -272,9 +272,7 @@ void middleman::start() {
}}; }};
backend().thread_id(thread_.get_id()); backend().thread_id(thread_.get_id());
} }
auto basp = named_broker<basp_broker>(atom("BASP")); // Default implementation of the stream server.
manager_ = make_middleman_actor(system(), basp);
// Install stream serv into the actor system.
class stream_serv : public raw_event_based_actor, class stream_serv : public raw_event_based_actor,
public detail::stream_multiplexer::backend { public detail::stream_multiplexer::backend {
public: public:
...@@ -354,11 +352,20 @@ void middleman::start() { ...@@ -354,11 +352,20 @@ void middleman::start() {
return result; return result;
} }
void on_exit() override {
// Make sure to not keep references to remotes after shutdown.
remotes().clear();
}
private: private:
detail::incoming_stream_multiplexer incoming_; detail::incoming_stream_multiplexer incoming_;
detail::outgoing_stream_multiplexer outgoing_; detail::outgoing_stream_multiplexer outgoing_;
}; };
auto ssi = system().spawn<stream_serv, lazy_init + hidden>(actor_cast<actor>(basp)); // Spawn utility actors.
auto basp = named_broker<basp_broker>(atom("BASP"));
manager_ = make_middleman_actor(system(), basp);
auto hdl = actor_cast<actor>(basp);
auto ssi = system().spawn<stream_serv, lazy_init + hidden>(std::move(hdl));
system().stream_serv(actor_cast<strong_actor_ptr>(std::move(ssi))); system().stream_serv(actor_cast<strong_actor_ptr>(std::move(ssi)));
} }
......
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