RxCpp and copy operations

Posted on 15 mins read

I’ve decided to investigate rxcpp library and get some better understanding of what is going on behind the scene with copy/move operations the library will do for the emitted objects.

For that I’ve written a simple hello world like app, just to get the ball rolling:

#include <rxcpp/rx.hpp>
#include <iostream>

int main()
{
    auto o = rxcpp::observable<>::from<int>(1,2,3);

    o.subscribe([](int item)
                { 
                    std::cout << item << std::endl; 
                });
    
    return 0;
}

When compiling and running (assuming rxcpp sources are in the include dirs and C++11 features set is ON), we get this output:

1
2
3

Ok, nothing surprising, knowing reactive source and subscription, that’s pretty much expected - an reactive equivalent of iterating over fixed set of values and printing them out.

Objects

Off course with primitive types like int there isn’t much to look after, but let’s switch to custom classes and let’s start to emit some objects ar the run-time.

#include <rxcpp/rx.hpp>
#include <iostream>

struct Foo
{
    int value;
};

int main()
{
    auto o = rxcpp::observable<>::from<int>(Foo{1}, Foo{2}, Foo{3});

    o.subscribe([](Foo item)
                { 
                    std::cout << item.value << std::endl; 
                });
    
    return 0;
}

We’ll get the same result, right? Right. At this point everything looks fairly simple, since we’re having an reactive equivalent of simple iterating over collection of Foo objects.

Now, let’s have a peek to what is happening between the point of where we’ve instantiated our Foo objects (1..3) to the point where we’re printing their values to the stdout.

To do so, let’s introduce:

  • a regular parametrised constructor
  • a copy constructor
  • a move constructor
  • a destructor

and let’s make each of them printing something meaningful to the console.

Out Foo class will look like this now:

struct Foo
{
    int value;
    
    Foo(int initial) : value (initial) { std::cout << "Foo: constructed from int " << initial << std::endl; }
    Foo(const Foo& other) : value(other.value) { std::cout << "Foo: COPY constructed from other Foo with " << other.value << std::endl; }
    Foo(Foo&& other) : value(other.value) { std::cout << "Foo: MOVE constructor for Foo with " << other.value <<std::endl; }
    ~Foo() { std::cout << "Foo: object with " << value << " destroyed " << std::endl; }
};

When we build and run our example, we get this:

Foo: constructed from int 3
Foo: constructed from int 2
Foo: constructed from int 1
Foo: COPY constructed from other Foo with 3
Foo: COPY constructed from other Foo with 2
Foo: COPY constructed from other Foo with 1
Foo: COPY constructed from other Foo with 1
Foo: COPY constructed from other Foo with 2
Foo: COPY constructed from other Foo with 3
Foo: MOVE constructor for Foo with 1
Foo: MOVE constructor for Foo with 2
Foo: MOVE constructor for Foo with 3
Foo: MOVE constructor for Foo with 1
Foo: MOVE constructor for Foo with 2
Foo: MOVE constructor for Foo with 3
Foo: MOVE constructor for Foo with 1
Foo: MOVE constructor for Foo with 2
Foo: MOVE constructor for Foo with 3
Foo: MOVE constructor for Foo with 1
Foo: MOVE constructor for Foo with 2
Foo: MOVE constructor for Foo with 3
Foo: object with 3 destroyed 
Foo: object with 2 destroyed 
Foo: object with 1 destroyed 
Foo: MOVE constructor for Foo with 1
Foo: MOVE constructor for Foo with 2
Foo: MOVE constructor for Foo with 3
Foo: MOVE constructor for Foo with 1
Foo: MOVE constructor for Foo with 2
Foo: MOVE constructor for Foo with 3
Foo: object with 3 destroyed 
Foo: object with 2 destroyed 
Foo: object with 1 destroyed 
Foo: object with 3 destroyed 
Foo: object with 2 destroyed 
Foo: object with 1 destroyed 
Foo: object with 3 destroyed 
Foo: object with 2 destroyed 
Foo: object with 1 destroyed 
Foo: MOVE constructor for Foo with 1
Foo: MOVE constructor for Foo with 2
Foo: MOVE constructor for Foo with 3
Foo: object with 3 destroyed 
Foo: object with 2 destroyed 
Foo: object with 1 destroyed 
Foo: object with 3 destroyed 
Foo: object with 2 destroyed 
Foo: object with 1 destroyed 
Foo: object with 3 destroyed 
Foo: object with 2 destroyed 
Foo: object with 1 destroyed 
Foo: MOVE constructor for Foo with 1
Foo: MOVE constructor for Foo with 2
Foo: MOVE constructor for Foo with 3
Foo: object with 3 destroyed 
Foo: object with 2 destroyed 
Foo: object with 1 destroyed 
Foo: object with 1 destroyed 
Foo: object with 2 destroyed 
Foo: object with 3 destroyed 
Foo: MOVE constructor for Foo with 1
Foo: MOVE constructor for Foo with 2
Foo: MOVE constructor for Foo with 3
Foo: object with 3 destroyed 
Foo: object with 2 destroyed 
Foo: object with 1 destroyed 
Foo: object with 1 destroyed 
Foo: object with 2 destroyed 
Foo: object with 3 destroyed 
Foo: COPY constructed from other Foo with 1
Foo: COPY constructed from other Foo with 2
Foo: COPY constructed from other Foo with 3
Foo: COPY constructed from other Foo with 1
Foo: COPY constructed from other Foo with 2
Foo: COPY constructed from other Foo with 3
Foo: COPY constructed from other Foo with 1
Foo: COPY constructed from other Foo with 2
Foo: COPY constructed from other Foo with 3
Foo: COPY constructed from other Foo with 1
Foo: COPY constructed from other Foo with 2
Foo: COPY constructed from other Foo with 3
Foo: COPY constructed from other Foo with 1
Foo: COPY constructed from other Foo with 2
Foo: COPY constructed from other Foo with 3
Foo: object with 3 destroyed 
Foo: object with 2 destroyed 
Foo: object with 1 destroyed 
Foo: object with 3 destroyed 
Foo: object with 2 destroyed 
Foo: object with 1 destroyed 
Foo: COPY constructed from other Foo with 1
Foo: COPY constructed from other Foo with 2
Foo: COPY constructed from other Foo with 3
Foo: object with 3 destroyed 
Foo: object with 2 destroyed 
Foo: object with 1 destroyed 
Foo: COPY constructed from other Foo with 1
Foo: COPY constructed from other Foo with 2
Foo: COPY constructed from other Foo with 3
Foo: COPY constructed from other Foo with 1
Foo: COPY constructed from other Foo with 2
Foo: COPY constructed from other Foo with 3
Foo: COPY constructed from other Foo with 1
Foo: COPY constructed from other Foo with 2
Foo: COPY constructed from other Foo with 3
Foo: object with 3 destroyed 
Foo: object with 2 destroyed 
Foo: object with 1 destroyed 
Foo: object with 3 destroyed 
Foo: object with 2 destroyed 
Foo: object with 1 destroyed 
Foo: COPY constructed from other Foo with 1
Foo: MOVE constructor for Foo with 1
1
Foo: object with 1 destroyed 
Foo: object with 1 destroyed 
Foo: COPY constructed from other Foo with 2
Foo: MOVE constructor for Foo with 2
2
Foo: object with 2 destroyed 
Foo: object with 2 destroyed 
Foo: COPY constructed from other Foo with 3
Foo: MOVE constructor for Foo with 3
3
Foo: object with 3 destroyed 
Foo: object with 3 destroyed 
Foo: object with 3 destroyed 
Foo: object with 2 destroyed 
Foo: object with 1 destroyed 
Foo: object with 3 destroyed 
Foo: object with 2 destroyed 
Foo: object with 1 destroyed 
Foo: object with 3 destroyed 
Foo: object with 2 destroyed 
Foo: object with 1 destroyed 
Foo: object with 3 destroyed 
Foo: object with 2 destroyed 
Foo: object with 1 destroyed 
Foo: object with 3 destroyed 
Foo: object with 2 destroyed 
Foo: object with 1 destroyed

Sweet lord!!!

Surprised? Well, I was.

It looks like there are plenty of things to start worrying about - or at least to think about when making use of rxcpp.

Things are probably acceptable when the objects passed around are as simple as our Foo is for the moment, but just imagine, what could be the consequences when the objects are large (take a lot of memory) and when constructing/copying and/or destroying is a long and costful operation.

Hey, it’s C++ with its object passing by value strategy. You’ve seen that before…

Before we analyse of what’s going on further down the road, we can already put a first conclusion here.

When you want to do reactive programming in C++ with rxcpp with concrete objects passed by, then you should always check whether you are prepared for a rather massive and not always clearly foreseeable amount of copying, construction and destructions going on, and make sure your classes are designed for that.

Let’s now simplify our example to having just one object emitted and let’s use our Foo tracing abilities to check what is happening on a minimum scale.

Our observable will emit just one Foo instance:

    auto o = rxcpp::observable<>::from<int>(Foo{1});

This gives us the following trace to the stdout:

Foo: constructed from int 1
Foo: COPY constructed from other Foo with 1
Foo: COPY constructed from other Foo with 1
Foo: MOVE constructor for Foo with 1
Foo: MOVE constructor for Foo with 1
Foo: MOVE constructor for Foo with 1
Foo: MOVE constructor for Foo with 1
Foo: object with 1 destroyed 
Foo: MOVE constructor for Foo with 1
Foo: MOVE constructor for Foo with 1
Foo: object with 1 destroyed 
Foo: object with 1 destroyed 
Foo: object with 1 destroyed 
Foo: MOVE constructor for Foo with 1
Foo: object with 1 destroyed 
Foo: object with 1 destroyed 
Foo: object with 1 destroyed 
Foo: MOVE constructor for Foo with 1
Foo: object with 1 destroyed 
Foo: object with 1 destroyed 
Foo: MOVE constructor for Foo with 1
Foo: object with 1 destroyed 
Foo: object with 1 destroyed 
Foo: COPY constructed from other Foo with 1
Foo: COPY constructed from other Foo with 1
Foo: COPY constructed from other Foo with 1
Foo: COPY constructed from other Foo with 1
Foo: COPY constructed from other Foo with 1
Foo: object with 1 destroyed 
Foo: object with 1 destroyed 
Foo: COPY constructed from other Foo with 1
Foo: object with 1 destroyed 
Foo: COPY constructed from other Foo with 1
Foo: COPY constructed from other Foo with 1
Foo: COPY constructed from other Foo with 1
Foo: object with 1 destroyed 
Foo: object with 1 destroyed 
Foo: COPY constructed from other Foo with 1
Foo: MOVE constructor for Foo with 1
1
Foo: object with 1 destroyed 
Foo: object with 1 destroyed 
Foo: object with 1 destroyed 
Foo: object with 1 destroyed 
Foo: object with 1 destroyed 
Foo: object with 1 destroyed 
Foo: object with 1 destroyed

Ok, easier to follow now.

First interesting bit is that after 1 has been put to the std::cout in the subscriber’s lambda, we still later see 7 instances of Foo destroyed. Wow. Sounds like a lot to me.

To be fair, our subscriber uses plain Foo as an argument, so at least one copy is due to argument passing by value there. Let’s change the subsriber’s lambda to use const reference, and see how that changes things.

o.subscribe([](const Foo& item)
                { 
                    std::cout << item.value << std::endl; 
                });

As expected, we get this:

...
1
Foo: object with 1 destroyed 
Foo: object with 1 destroyed 
Foo: object with 1 destroyed 
Foo: object with 1 destroyed 
Foo: object with 1 destroyed 
Foo: object with 1 destroyed

OK, we’ve saved one, where we could.

Let’s see how that the picture with one instance being emitted changes after we introduce some additional operators between the from observable source and the subscriber.

Let’s add take(1), which shuld be logicall invariant here, and let’s see what happens.

With:

	auto o = rxcpp::observable<>::from<Foo>(Foo{1})
			.take(1);

We’re getting this:

...
1
Foo: object with 1 destroyed 
Foo: object with 1 destroyed 
Foo: object with 1 destroyed 
Foo: object with 1 destroyed 
Foo: object with 1 destroyed 
Foo: object with 1 destroyed 
Foo: object with 1 destroyed 
Foo: object with 1 destroyed 
Foo: object with 1 destroyed

Ok, so now, after adding just one operator we’re having +3 Foo instances living somewhere in the pipeline.

Let’s continue with an extra filter operator (this time also using const reference in the lambda, to make sure we don’t introduce extra copying on argument passing there):

    auto o = rxcpp::observable<>::from<Foo>(Foo{1})
			.take(1)
			.filter([](const Foo& foo){ return true; });

The results are:

...
1
Foo: object with 1 destroyed 
Foo: object with 1 destroyed 
Foo: object with 1 destroyed 
Foo: object with 1 destroyed 
Foo: object with 1 destroyed 
Foo: object with 1 destroyed 
Foo: object with 1 destroyed 
Foo: object with 1 destroyed 
Foo: object with 1 destroyed 
Foo: object with 1 destroyed

Ok, this resulted with +1 extra instance. Let’s give another logical invariant operation: skip(0) to the process:

    auto o = rxcpp::observable<>::from<Foo>(Foo{1})
			.take(1)
			.filter([](const Foo& foo){ return true; })
			.skip(0);

results are:

1
Foo: object with 1 destroyed 
Foo: object with 1 destroyed 
Foo: object with 1 destroyed 
Foo: object with 1 destroyed 
Foo: object with 1 destroyed 
Foo: object with 1 destroyed 
Foo: object with 1 destroyed 
Foo: object with 1 destroyed 
Foo: object with 1 destroyed 
Foo: object with 1 destroyed 
Foo: object with 1 destroyed 
Foo: object with 1 destroyed 
Foo: object with 1 destroyed

Now, +3 extra instances.

Hmm. The next conclusion emerges.

It looks like it is quite hard to predict the exact influence on number of copings/moves that happen when introducing additional reactive operators.

Shared pointers

We could avoid all those problematic behaviours by dropping by-value passing and by switching to shared pointers instead.

Let’s change the code to use dynamically allocated objects and std::shared_ptr to control it’s ownership and lifetime.

#include <rxcpp/rx.hpp>
#include <iostream>
#include <memory>

struct Foo
{
    int value;
    
    Foo(int initial) : value (initial) { std::cout << "Foo: constructed from int " << initial << std::endl; }
    Foo(const Foo& other) : value(other.value) { std::cout << "Foo: COPY constructed from other Foo with " << other.value << std::endl; }
    Foo(Foo&& other) : value(other.value) { std::cout << "Foo: MOVE constructor for Foo with " << other.value <<std::endl; }
    ~Foo() { std::cout << "Foo: object with " << value << " destroyed " << std::endl; }
};

int main()
{
    auto o = rxcpp::observable<>::from<std::shared_ptr<Foo>>(std::make_shared<Foo>(1),std::make_shared<Foo>(2),std::make_shared<Foo>(3))
			.take(3)
			.filter([](std::shared_ptr<Foo> foo){ return true; })
			.skip(0);

    o.subscribe([](std::shared_ptr<Foo> item)
                { 
                    std::cout << item->value << std::endl; 
                });
    
    return 0;
}

Ok, iterating over three objects dyanamically created gives the following trace now:

Foo: constructed from int 3
Foo: constructed from int 2
Foo: constructed from int 1
1
2
3
Foo: object with 3 destroyed 
Foo: object with 2 destroyed 
Foo: object with 1 destroyed

How beautiful! And actually as executed, since all copying/moves are happening now on the smart pointer objects rather than the pointee (Foo instance).

One more thing, when looking to the trace above (as well as the ones from the section before), you see that the iteration 1..3 is happening in the expected order, but the object construction is in reverse (Foo(3) is constructed first). That is because of how Observable::from source is implemented. It is a variadic template function and I expect that it’s been calling recursively template specialisations, unwinding the tail element first at each recursion level.

Surprisingly, destruction sequence is not a perfect mirror of it. I would personally find more intuitive if objects constructed first are released as last, but that’s not the case here. Could that be a problem in a real-life scenario? Probably not.

Ok, time for even deeper investigation. Let’s imaging we add another map operator that converts from Foo to a new type of object - Bar somewhere in the pipeline.

Let’s add Bar class that can be instantiated from Foo and carry on it’s integer value. The example code would look like this:

#include <rxcpp/rx.hpp>
#include <iostream>
#include <memory>

struct Foo
{
    int value;
    
    Foo(int initial) : value (initial) { std::cout << "Foo: constructed from int " << initial << std::endl; }
    Foo(const Foo& other) : value(other.value) { std::cout << "Foo: COPY constructed from other Foo with " << other.value << std::endl; }
    Foo(Foo&& other) : value(other.value) { std::cout << "Foo: MOVE constructor for Foo with " << other.value <<std::endl; }
    ~Foo() { std::cout << "Foo: object with " << value << " destroyed " << std::endl; }
};

struct Bar
{
	int value;
	
    Bar(const Foo& src) : value (src.value) { std::cout << "Bar: constructed from Foo with " << src.value << std::endl; }
    Bar(const Bar& other) : value(other.value) { std::cout << "Bar: COPY constructed from other Bar with " << other.value << std::endl; }
    Bar(Bar&& other) : value(other.value) { std::cout << "Bar: MOVE constructor for Bar with " << other.value <<std::endl; }
    ~Bar() { std::cout << "Bar: object with " << value << " destroyed " << std::endl; }
};

int main()
{
    auto o = rxcpp::observable<>::from<std::shared_ptr<Foo>>(std::make_shared<Foo>(1),std::make_shared<Foo>(2),std::make_shared<Foo>(3))
			.take(3)
			.filter([](std::shared_ptr<Foo> foo){ return true; })
			.map([](std::shared_ptr<Foo> foo){ return std::make_shared<Bar>(*foo); })
			.skip(0);

    o.subscribe([](std::shared_ptr<Bar> item)
                { 
                    std::cout << item->value << std::endl; 
                });
    
    return 0;
}

The results of this program execution is:

Foo: constructed from int 3
Foo: constructed from int 2
Foo: constructed from int 1
Bar: constructed from Foo with 1
1
Bar: object with 1 destroyed 
Bar: constructed from Foo with 2
2
Bar: object with 2 destroyed 
Bar: constructed from Foo with 3
3
Bar: object with 3 destroyed 
Foo: object with 3 destroyed 
Foo: object with 2 destroyed 
Foo: object with 1 destroyed

The interesting bit is that the Bar objects have short lifetime, only from the moment when they are constructed in the pipeline (map operator) until the point when they have been consumed by the subscriber. That is nice and well expected, since reactive programming is indeed about processing data as the they arrive, one at a time.

A bit surprise on that the Foo objects live longer, until entire processing is complete. This might be explain by probably having an internal list of shared pointers the Observable::from implementation is keeping for it’s lifetime.

Perhaps rxcpp could be improved in this place, so that when particular elements are emitted, elements can be removed from that inner list and destructors of Foo are called immediately after, e.g. like this:

Foo: constructed from int 3
Foo: constructed from int 2
Foo: constructed from int 1
Bar: constructed from Foo with 1
1
Bar: object with 1 destroyed 
Foo: object with 1 destroyed 	<-- expected here
Bar: constructed from Foo with 2
2
Bar: object with 2 destroyed 
Foo: object with 2 destroyed 	<-- expected here
Bar: constructed from Foo with 3
3
Bar: object with 3 destroyed 
Foo: object with 3 destroyed 	<!-- no change

That would give IMO more natural behaviour (as for reactive data processing), with as little kept in memory at a time as necessary to proceed.

Final conclusion

Use dynamic allocation and smart pointers when using rxcpp to process non-primitive data types, unless you really have a good reason and time to investigate exact impact on your classes lifetime, resource consumption to do it differently.

comments powered by Disqus