Open simonvpe opened 6 years ago
Yes, there are many circular references in rxcpp. These are broken by the unsubscribe signal. Is there an expression that actually demonstrates a leak?
After producing a MCVE I'm not sure it is take
that is the problem anymore.
#include <rxcpp/rx.hpp>
#include <iostream>
struct foo : public rxcpp::sources::source_base<int> {
static int refcount;
explicit foo() { std::cout << "++refcount = " << ++refcount << '\n'; }
~foo() { std::cout << "--refcount = " << --refcount << '\n'; }
foo(const foo &other) { std::cout << "++refcount = " << ++refcount << '\n'; }
foo(foo &&) { std::cout << "++refcount = " << ++refcount << '\n'; }
foo &operator=(foo &&) = default;
foo &operator=(const foo &) = default;
template <typename Subscriber>
void on_subscribe(Subscriber subscriber) const {
for (auto i = 0; i < 100; ++i) {
subscriber.on_next(i);
}
subscriber.on_completed();
}
};
auto create_foo_observable() {
return rxcpp::observable<int, foo>{foo{}}
.lift<int>([](rxcpp::subscriber<int> subscriber) {
subscriber.get_subscription().add([] {
std::cout << "Unsubscribed (refcount = " << foo::refcount << ")\n";
});
return subscriber;
})
.as_dynamic();
}
int foo::refcount = 0;
int main() {
{
create_foo_observable()
.observe_on(rxcpp::observe_on_new_thread())
.take(10)
.as_blocking()
.subscribe([](auto) {});
}
std::cout << "Final refcount is " << foo::refcount << '\n';
}
++refcount = 1
++refcount = 2
++refcount = 3
++refcount = 4
++refcount = 5
--refcount = 4
--refcount = 3
++refcount = 4
++refcount = 5
++refcount = 6
++refcount = 7
--refcount = 6
--refcount = 5
--refcount = 4
--refcount = 3
--refcount = 2
--refcount = 1
Unsubscribed (refcount = 1)
Final refcount is 1
If I remove .observe_on(rxcpp::observe_on_new_thread())
the final refcount is 0. Maybe this is not how you are supposed to create your own observables? In my real code I start up an async process using boost::asio
that calls the methods on the subscriber, but the general boilerplate is about the same.
If I use rxcpp::observable<>::create()
the result is more pleasing however.
#include <rxcpp/rx.hpp>
struct bar {
static int refcount;
explicit bar() { std::cout << "++refcount = " << ++refcount << '\n'; }
~bar() { std::cout << "--refcount = " << --refcount << '\n'; }
bar(const bar &other) { std::cout << "++refcount = " << ++refcount << '\n'; }
bar(bar &&) { std::cout << "++refcount = " << ++refcount << '\n'; }
bar &operator=(bar &&) = default;
bar &operator=(const bar &) = default;
};
int bar::refcount = 0;
int main() {
{
rxcpp::observable<>::create<int>([state = bar{}](auto subscriber) {
for (auto i = 0; i < 100; ++i) {
subscriber.on_next(i);
}
subscriber.on_completed();
})
.observe_on(rxcpp::observe_on_new_thread())
.take(10)
.as_blocking()
.subscribe([](auto) {});
}
// <- Need to wait for cleanup or `bar::refcount == 1`
using namespace std::chrono_literals;
std::this_thread::sleep_for(100ms);
std::cout << "Final refcount is " << bar::refcount << '\n';
}
++refcount = 1
++refcount = 2
++refcount = 3
++refcount = 4
++refcount = 5
--refcount = 4
--refcount = 3
--refcount = 2
++refcount = 3
++refcount = 4
++refcount = 5
--refcount = 4
--refcount = 3
++refcount = 4
++refcount = 5
++refcount = 6
--refcount = 5
++refcount = 6
--refcount = 5
--refcount = 4
++refcount = 5
++refcount = 6
--refcount = 5
++refcount = 6
--refcount = 5
--refcount = 4
--refcount = 3
--refcount = 2
--refcount = 1
--refcount = 0
Final refcount is 0
I expect the the first will get to zero if the sleep in the second is added before printing the final refcount.
The refcount waits for the observe_on to shutdown.
There is a circular dependency here. I'll try to figure out how to resolve it myself I presume. It would be a good idea to run the tests with valgrind or sanitizers though, since a lot of issues are about memory leaks.