Commit 8a885b53 authored by Dominik Charousset's avatar Dominik Charousset

Fix memory leak in buffered stages

parent eacf4343
...@@ -651,7 +651,7 @@ public: ...@@ -651,7 +651,7 @@ public:
pull(batch_size - buf_.size()); pull(batch_size - buf_.size());
auto n = std::min(max_demand_, buf_.size()); auto n = std::min(max_demand_, buf_.size());
if (n == 0) if (n == 0)
break; return;
batch_.assign(std::make_move_iterator(buf_.begin()), batch_.assign(std::make_move_iterator(buf_.begin()),
std::make_move_iterator(buf_.begin() + n)); std::make_move_iterator(buf_.begin() + n));
buf_.erase(buf_.begin(), buf_.begin() + n); buf_.erase(buf_.begin(), buf_.begin() + n);
...@@ -666,7 +666,8 @@ public: ...@@ -666,7 +666,8 @@ public:
for (auto& out : outputs_) for (auto& out : outputs_)
out.sink.on_complete(); out.sink.on_complete();
outputs_.clear(); outputs_.clear();
break; do_on_complete();
return;
} }
} }
} }
...@@ -790,11 +791,11 @@ public: ...@@ -790,11 +791,11 @@ public:
} }
void ref_disposable() const noexcept override { void ref_disposable() const noexcept override {
return super::ref_disposable(); this->ref();
} }
void deref_disposable() const noexcept override { void deref_disposable() const noexcept override {
return super::ref_disposable(); this->deref();
} }
// -- implementation of observable<T>::impl ---------------------------------- // -- implementation of observable<T>::impl ----------------------------------
...@@ -827,11 +828,17 @@ public: ...@@ -827,11 +828,17 @@ public:
void on_next(span<const In> items) final { void on_next(span<const In> items) final {
CAF_ASSERT(in_flight_ >= items.size()); CAF_ASSERT(in_flight_ >= items.size());
if (!this->completed_) {
in_flight_ -= items.size(); in_flight_ -= items.size();
do_on_next(items); if (!do_on_next(items)) {
this->try_push();
shutdown();
} else {
this->try_push(); this->try_push();
try_fetch_more(); try_fetch_more();
} }
}
}
void on_complete() override { void on_complete() override {
sub_ = nullptr; sub_ = nullptr;
...@@ -879,7 +886,7 @@ private: ...@@ -879,7 +886,7 @@ private:
} }
/// Transforms input items to outputs. /// Transforms input items to outputs.
virtual void do_on_next(span<const In> items) = 0; virtual bool do_on_next(span<const In> items) = 0;
}; };
/// Broadcasts its input to all observers without modifying it. /// Broadcasts its input to all observers without modifying it.
...@@ -891,8 +898,9 @@ public: ...@@ -891,8 +898,9 @@ public:
using super::super; using super::super;
private: private:
void do_on_next(span<const T> items) override { bool do_on_next(span<const T> items) override {
this->append_to_buf(items.begin(), items.end()); this->append_to_buf(items.begin(), items.end());
return true;
} }
}; };
...@@ -940,14 +948,15 @@ public: ...@@ -940,14 +948,15 @@ public:
std::tuple<Step, Steps...> steps; std::tuple<Step, Steps...> steps;
private: private:
void do_on_next(span<const input_type> items) override { bool do_on_next(span<const input_type> items) override {
auto f = [this, items](auto& step, auto&... steps) { auto f = [this, items](auto& step, auto&... steps) {
term_step<output_type> term{this}; term_step<output_type> term{this};
for (auto&& item : items) for (auto&& item : items)
if (!step.on_next(item, steps..., term)) if (!step.on_next(item, steps..., term))
return; return false;
return true;
}; };
std::apply(f, steps); return std::apply(f, steps);
} }
void do_on_complete() override { void do_on_complete() override {
...@@ -1025,6 +1034,7 @@ public: ...@@ -1025,6 +1034,7 @@ public:
auto pimpl = make_counted<impl>(source_.ptr()->ctx(), std::move(steps_)); auto pimpl = make_counted<impl>(source_.ptr()->ctx(), std::move(steps_));
auto res = pimpl->as_observable(); auto res = pimpl->as_observable();
source_.subscribe(observer<input_type>{std::move(pimpl)}); source_.subscribe(observer<input_type>{std::move(pimpl)});
source_ = nullptr;
return res; return res;
} }
......
...@@ -69,6 +69,7 @@ SCENARIO("for_each iterates all values in a stream") { ...@@ -69,6 +69,7 @@ SCENARIO("for_each iterates all values in a stream") {
CHECK_EQ(inputs, outputs); CHECK_EQ(inputs, outputs);
} }
/* subtest */ { /* subtest */ {
auto completed = false;
auto inputs = std::vector<int>{21, 21, 21, 21, 21, 21, 21}; auto inputs = std::vector<int>{21, 21, 21, 21, 21, 21, 21};
auto outputs = std::vector<int>{}; auto outputs = std::vector<int>{};
ctx->make_observable() ctx->make_observable()
...@@ -76,8 +77,13 @@ SCENARIO("for_each iterates all values in a stream") { ...@@ -76,8 +77,13 @@ SCENARIO("for_each iterates all values in a stream") {
.as_observable() .as_observable()
.take(7) .take(7)
.map([](int x) { return x * 3; }) .map([](int x) { return x * 3; })
.for_each([&outputs](int x) { outputs.emplace_back(x); }); .for_each([&outputs](int x) { outputs.emplace_back(x); },
[](const error& reason) {
FAIL("on_error called: " << reason);
},
[&completed] { completed = true; });
ctx->run(); ctx->run();
CHECK(completed);
CHECK_EQ(inputs, outputs); CHECK_EQ(inputs, outputs);
} }
} }
......
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