Commit 5477e042 authored by Dominik Charousset's avatar Dominik Charousset

Fix memory leaks

parent 55801c50
...@@ -147,6 +147,13 @@ public: ...@@ -147,6 +147,13 @@ public:
observable& operator=(observable&&) noexcept = default; observable& operator=(observable&&) noexcept = default;
observable& operator=(const observable&) noexcept = default; observable& operator=(const observable&) noexcept = default;
void dispose() {
if (pimpl_) {
pimpl_->dispose();
pimpl_ = nullptr;
}
}
disposable as_disposable() const& noexcept { disposable as_disposable() const& noexcept {
return disposable{pimpl_}; return disposable{pimpl_};
} }
...@@ -693,6 +700,7 @@ public: ...@@ -693,6 +700,7 @@ public:
step.on_complete(steps..., term_); step.on_complete(steps..., term_);
}; };
std::apply(f, steps_); std::apply(f, steps_);
sub_ = nullptr;
} }
} }
...@@ -702,6 +710,7 @@ public: ...@@ -702,6 +710,7 @@ public:
step.on_error(what, steps..., term_); step.on_error(what, steps..., term_);
}; };
std::apply(f, steps_); std::apply(f, steps_);
sub_ = nullptr;
} }
} }
...@@ -2158,11 +2167,15 @@ public: ...@@ -2158,11 +2167,15 @@ public:
// -- implementation of disposable::impl ------------------------------------- // -- implementation of disposable::impl -------------------------------------
void dispose() override { void dispose() override {
// nop if (!disposed_) {
disposed_ = true;
for (auto& obs : observers_)
obs.on_complete();
}
} }
bool disposed() const noexcept override { bool disposed() const noexcept override {
return true; return disposed_;
} }
void ref_disposable() const noexcept override { void ref_disposable() const noexcept override {
...@@ -2188,11 +2201,20 @@ public: ...@@ -2188,11 +2201,20 @@ public:
} }
disposable subscribe(observer<output_type> sink) override { disposable subscribe(observer<output_type> sink) override {
return this->do_subscribe(sink.ptr()); if (!disposed_) {
auto ptr = make_counted<subscription::nop_impl>();
sink.ptr()->on_subscribe(subscription{std::move(ptr)});
observers_.emplace_back(sink);
return sink.as_disposable();
} else {
return this->reject_subscription(sink, sec::disposed);
}
} }
private: private:
coordinator* ctx_; coordinator* ctx_;
bool disposed_ = false;
std::vector<observer<output_type>> observers_;
}; };
/// An observable with minimal internal logic. Useful for writing unit tests. /// An observable with minimal internal logic. Useful for writing unit tests.
......
...@@ -433,6 +433,7 @@ public: ...@@ -433,6 +433,7 @@ public:
size_t on_cancel(observer_impl_t* sink) { size_t on_cancel(observer_impl_t* sink) {
if (auto i = find(sink); i != outputs_.end()) { if (auto i = find(sink); i != outputs_.end()) {
outputs_.erase(i); outputs_.erase(i);
// TODO: shut down on last cancel?
push(); push();
return next_demand(); return next_demand();
} else { } else {
......
...@@ -33,6 +33,28 @@ public: ...@@ -33,6 +33,28 @@ public:
void dispose() final; void dispose() final;
}; };
/// A trivial subscription type that drops all member function calls.
class CAF_CORE_EXPORT nop_impl final : public ref_counted,
public subscription::impl {
public:
// -- friends --------------------------------------------------------------
CAF_INTRUSIVE_PTR_FRIENDS(nop_impl)
bool disposed() const noexcept override;
void ref_disposable() const noexcept override;
void deref_disposable() const noexcept override;
void cancel() override;
void request(size_t n) override;
private:
bool disposed_ = false;
};
// -- constructors, destructors, and assignment operators -------------------- // -- constructors, destructors, and assignment operators --------------------
explicit subscription(intrusive_ptr<impl> pimpl) noexcept explicit subscription(intrusive_ptr<impl> pimpl) noexcept
......
...@@ -14,4 +14,24 @@ void subscription::impl::dispose() { ...@@ -14,4 +14,24 @@ void subscription::impl::dispose() {
cancel(); cancel();
} }
bool subscription::nop_impl::disposed() const noexcept {
return disposed_;
}
void subscription::nop_impl::ref_disposable() const noexcept {
ref();
}
void subscription::nop_impl::deref_disposable() const noexcept {
deref();
}
void subscription::nop_impl::cancel() {
disposed_ = true;
}
void subscription::nop_impl::request(size_t) {
// nop
}
} // namespace caf::flow } // namespace caf::flow
...@@ -25,6 +25,33 @@ struct fixture : test_coordinator_fixture<> { ...@@ -25,6 +25,33 @@ struct fixture : test_coordinator_fixture<> {
BEGIN_FIXTURE_SCOPE(fixture) BEGIN_FIXTURE_SCOPE(fixture)
SCENARIO("a broadcaster pushes items to single subscribers") {
GIVEN("a broadcaster with one source and one sink") {
auto uut = flow::make_broadcaster_impl<int>(ctx.get());
auto src = flow::make_passive_observable<int>(ctx.get());
auto snk = flow::make_passive_observer<int>();
src->subscribe(uut->as_observer());
uut->subscribe(snk->as_observer());
WHEN("the source emits 10 items") {
THEN("the broadcaster forwards them to its sink") {
snk->sub.request(13);
ctx->run();
CHECK_EQ(src->demand, 13u);
snk->sub.request(7);
ctx->run();
CHECK_EQ(src->demand, 20u);
auto inputs = std::vector<int>{1, 2, 3, 4, 5, 6, 7, 8, 9, 10};
src->push(make_span(inputs));
CHECK_EQ(src->demand, 10u);
CHECK_EQ(uut->buffered(), 0u);
CHECK_EQ(snk->buf, std::vector<int>({1, 2, 3, 4, 5, 6, 7, 8, 9, 10}));
src->complete();
ctx->run();
}
}
}
}
SCENARIO("a broadcaster pushes items to all subscribers at the same time") { SCENARIO("a broadcaster pushes items to all subscribers at the same time") {
GIVEN("a broadcaster with one source and three sinks") { GIVEN("a broadcaster with one source and three sinks") {
auto uut = flow::make_broadcaster_impl<int>(ctx.get()); auto uut = flow::make_broadcaster_impl<int>(ctx.get());
...@@ -64,6 +91,8 @@ SCENARIO("a broadcaster pushes items to all subscribers at the same time") { ...@@ -64,6 +91,8 @@ SCENARIO("a broadcaster pushes items to all subscribers at the same time") {
snk2->sub.request(14); snk2->sub.request(14);
ctx->run(); ctx->run();
CHECK_EQ(src->demand, 18u); CHECK_EQ(src->demand, 18u);
src->complete();
ctx->run();
} }
} }
} }
......
...@@ -52,6 +52,8 @@ SCENARIO("concatenate processes inputs sequentially") { ...@@ -52,6 +52,8 @@ SCENARIO("concatenate processes inputs sequentially") {
ctx->run(); ctx->run();
CHECK_EQ(snk->state, flow::observer_state::subscribed); CHECK_EQ(snk->state, flow::observer_state::subscribed);
CHECK(snk->buf.empty()); CHECK(snk->buf.empty());
uut->dispose();
ctx->run();
} }
} }
} }
......
...@@ -52,6 +52,8 @@ SCENARIO("mergers round-robin over their inputs") { ...@@ -52,6 +52,8 @@ SCENARIO("mergers round-robin over their inputs") {
ctx->run(); ctx->run();
CHECK_EQ(snk->state, flow::observer_state::subscribed); CHECK_EQ(snk->state, flow::observer_state::subscribed);
CHECK(snk->buf.empty()); CHECK(snk->buf.empty());
uut->dispose();
ctx->run();
} }
} }
} }
......
...@@ -22,10 +22,10 @@ struct fixture : test_coordinator_fixture<> { ...@@ -22,10 +22,10 @@ struct fixture : test_coordinator_fixture<> {
BEGIN_FIXTURE_SCOPE(fixture) BEGIN_FIXTURE_SCOPE(fixture)
SCENARIO("a mute observable never invokes any callbacks") { SCENARIO("a mute observable never invokes any callbacks except when disposed") {
GIVEN("an never<int32>") { GIVEN("a never<int32>") {
WHEN("an observer subscribes") { WHEN("an observer subscribes") {
THEN("the observer never observes any activity") { THEN("the observer never receives any events") {
auto uut = ctx->make_observable().never<int32_t>(); auto uut = ctx->make_observable().never<int32_t>();
auto snk = flow::make_passive_observer<int32_t>(); auto snk = flow::make_passive_observer<int32_t>();
uut.subscribe(snk->as_observer()); uut.subscribe(snk->as_observer());
...@@ -39,6 +39,28 @@ SCENARIO("a mute observable never invokes any callbacks") { ...@@ -39,6 +39,28 @@ SCENARIO("a mute observable never invokes any callbacks") {
} }
} }
} }
GIVEN("a never<int32> that gets disposed") {
WHEN("an observer subscribes") {
THEN("the observer receives on_complete") {
auto uut = ctx->make_observable().never<int32_t>();
auto snk1 = flow::make_passive_observer<int32_t>();
auto snk2 = flow::make_passive_observer<int32_t>();
uut.subscribe(snk1->as_observer());
ctx->run();
if (CHECK(snk1->sub)) {
snk1->sub.request(42);
ctx->run();
CHECK_EQ(snk1->state, flow::observer_state::subscribed);
CHECK(snk1->buf.empty());
uut.dispose();
ctx->run();
CHECK_EQ(snk1->state, flow::observer_state::completed);
uut.subscribe(snk2->as_observer());
CHECK_EQ(snk2->state, flow::observer_state::aborted);
}
}
}
}
} }
END_FIXTURE_SCOPE() END_FIXTURE_SCOPE()
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