New IntraProcessManager capable of storing shared_ptr<const T>

[ivanpauno]

This is an enhancement of the IntraProcessManager, that allows storing both unique_ptr and shared_ptr. Design draft here.
It shows a better performance, specially in the case of many subscriptions sharing ownership of the same const message. Document with performance results here.

List of changes

• MappedRingBuffer can store both kind of smart pointers. push_and_replace, get, and pop methods for both signature.
• MappedRingBuffer updated, covering the new implementation.
• Added store_intra_process_message and take_intra_process_message methods with both signatures to the IntraProcessManager.
• Modified Publisher publish methods to use this new IntraProcessManager.
• Modified Subscription handle_intra_process_message method.
• Modified AnySubscriptionCallback dispatch_intra_process method.
• Stopped bounding intra-process related methods from SubscriptionFactory and PublisherFactory in Subscription and Publisher class.
• Split publisher.h and subscriber.h in two files, one with the base class and the other with the derived.
  The idea is to avoid a circular inclusion with intra_process_manager.h

To do

• The intra_process_manager test has been corrected only to make it pass, because of the api changes. But it's actually not covering all the new IPM methods (which are well tested at mapped_ring_buffer level).

[mjcarroll]

Partial Review

[mjcarroll]

CI through rclcpp:

• Linux
• Linux-aarch64
• macOS
• Windows

[mjcarroll]

@ivanpauno Looks to be some warnings on Linux and a failure on macOS, can you get those resolved?

[ivanpauno]

CI through rclcpp:

• Linux
• Linux-aarch64
• macOS
• Windows

[ivanpauno]

@ivanpauno Looks to be some warnings on Linux and a failure on macOS, can you get those resolved?

I still have an obscure warning on Windows. I will try to figure it out on Monday.

[ivanpauno]

CI through rclcpp:

• Windows

[mjcarroll]

LGTM, but would really appreciate another look from @wjwwood if you have the chance.

[wjwwood]

I'll try to have a look today.

[ivanpauno]

@wjwwood I will explain here why I split publisher.hpp and subscriber.hpp. Sorry for the long story.

Before we had only one store_intra_process_message_ method, which was bounded.

rclcpp/rclcpp/include/rclcpp/publisher.hpp

Line 189 in 6b10841

StoreMessageCallbackT store_intra_process_message_;

rclcpp/rclcpp/include/rclcpp/publisher.hpp

Line 165 in 6b10841

using StoreMessageCallbackT = std::function<uint64_t(uint64_t, void *, const std::type_info &)>;

The callback took a void *, so it was a member of PublisherBase, and it was bounded from 'nodes_interfaces::NodeTopics. The method definition was inside PublisherFactory`

rclcpp/rclcpp/src/rclcpp/node_interfaces/node_topics.cpp

Lines 45 to 59 in 6b10841

	if (use_intra_process) {
	auto context = node_base_->get_context();
	// Get the intra process manager instance for this context.
	auto ipm = context->get_sub_context<rclcpp::intra_process_manager::IntraProcessManager>();
	// Register the publisher with the intra process manager.
	uint64_t intra_process_publisher_id =
	publisher_factory.add_publisher_to_intra_process_manager(ipm.get(), publisher);
	// Create a function to be called when publisher to do the intra process publish.
	auto shared_publish_callback = publisher_factory.create_shared_publish_callback(ipm);
	publisher->setup_intra_process(
	intra_process_publisher_id,
	shared_publish_callback,
	ipm,
	publisher_options);
	}

With the idea of having two store_intra_process_message methods, one taking a unique_ptr and one taking a shared_ptr, I moved those methods to the publisher and I stopped bounding them.
Those methods call the equivalent store_intra_process_message methods in the IPM, so I needed to include it.
As the IPM was including publisher.hpp but it didn't need more than PublisherBase, I split it in two files and included publisher_base.hpp for avoiding a circular inclusion. I still needed to forward declare the IPM in publisher_base, as I let setup_intra_process method and a weak_ptr to the IPM there.
The only place where the IPM needed Publisher and not PublisherBase was here:

rclcpp/rclcpp/include/rclcpp/intra_process_manager.hpp

Lines 187 to 201 in 6b10841

	template<typename MessageT, typename Alloc>
	uint64_t
	add_publisher(
	typename Publisher<MessageT, Alloc>::SharedPtr publisher,
	size_t buffer_size = 0)
	{
	auto id = IntraProcessManager::get_next_unique_id();
	size_t size = buffer_size > 0 ? buffer_size : publisher->get_queue_size();
	auto mrb = mapped_ring_buffer::MappedRingBuffer<
	MessageT,
	typename Publisher<MessageT, Alloc>::MessageAlloc
	>::make_shared(size, publisher->get_allocator());
	impl_->add_publisher(id, publisher, mrb, size);
	return id;
	}

So I moved the MappedRingBuffer creation to the publisher.

In subscription.hpp, when I replaced get_intra_process_message_callback_ bounded method with two take_intra_process_message methods I had the same problem. So I applied a similar logic.

This could have been solved in another way: The get_intra_process_message_callback_ in Subscription already took a typed unique_ptr, the only difference with Publisher was that it was bounded from setup_intra_process method inside the factory.

rclcpp/rclcpp/include/rclcpp/subscription_factory.hpp

Lines 118 to 172 in 6b10841

	factory.setup_intra_process =
	[message_alloc](
	rclcpp::intra_process_manager::IntraProcessManager::SharedPtr ipm,
	rclcpp::SubscriptionBase::SharedPtr subscription,
	const rcl_subscription_options_t & subscription_options)
	{
	rclcpp::intra_process_manager::IntraProcessManager::WeakPtr weak_ipm = ipm;
	uint64_t intra_process_subscription_id = ipm->add_subscription(subscription);
	auto intra_process_options = rcl_subscription_get_default_options();
	intra_process_options.allocator = rclcpp::allocator::get_rcl_allocator<CallbackMessageT>(
	*message_alloc.get());
	intra_process_options.qos = subscription_options.qos;
	intra_process_options.ignore_local_publications = false;
	// function that will be called to take a MessageT from the intra process manager
	auto take_intra_process_message_func =
	[weak_ipm](
	uint64_t publisher_id,
	uint64_t message_sequence,
	uint64_t subscription_id,
	typename rclcpp::Subscription<CallbackMessageT, Alloc>::MessageUniquePtr & message)
	{
	auto ipm = weak_ipm.lock();
	if (!ipm) {
	// TODO(wjwwood): should this just return silently? Or return with a logged warning?
	throw std::runtime_error(
	"intra process take called after destruction of intra process manager");
	}
	ipm->take_intra_process_message<CallbackMessageT, Alloc>(
	publisher_id, message_sequence, subscription_id, message);
	};
	// function that is called to see if the publisher id matches any local publishers
	auto matches_any_publisher_func =
	[weak_ipm](const rmw_gid_t * sender_gid) -> bool
	{
	auto ipm = weak_ipm.lock();
	if (!ipm) {
	throw std::runtime_error(
	"intra process publisher check called "
	"after destruction of intra process manager");
	}
	return ipm->matches_any_publishers(sender_gid);
	};
	auto typed_sub_ptr = std::dynamic_pointer_cast<SubscriptionT>(subscription);
	typed_sub_ptr->setup_intra_process(
	intra_process_subscription_id,
	take_intra_process_message_func,
	matches_any_publisher_func,
	weak_ipm,
	intra_process_options
	);
	};

So I could bound the 'store' and 'take' methods to Publisher and Subscription with both shared_ptr and unique_ptr signature in a similar way, and avoid moving so many things.

I started splitting the publisher.hpp file, and when I realized of this last option I felt that it was more "natural" to don't have bounded methods.

Please if someone prefer the last option I commented or have a third one, let me know. This changes were only a "side effect", and it's not the main intention of this PR.

[ivanpauno]

According to this discussion, the comment here, #640, and the CppCoreGuideline (R.32 to R.36) I changed all the methods (that are subject of this PR) with unique_ptr or shared_ptr signature to agree with them.
I know that I will probably break code in other repositories (demos, test_rclcpp, tlsf_cpp, ...), specially because now std::move is explicitly needed when publishing a unique_ptr message.

@wjwwood @mjcarroll Let me know if we want to include the change of:
publish(std::unique_ptr<MessageT, MessageDeleter> & msg) to
publish(std::unique_ptr<MessageT, MessageDeleter> msg)
in this PR or not. In case that we want to include it, I will start opening the remaining PRs on Monday.

I think that the other open discussion is if we want to split the publisher.hpp and subscription.hpp files or not. Please, check my previous comment when possible.

Edit:
Another open discussion #690 (comment).

[wjwwood]

Just two thoughts while we continue to discuss the publish signatures we provide and how they interact with the IPM (some online and offline discussions).

[ivanpauno]

I will deprecate the shared_ptr publish method in another PR, opening all the needed PRs in other repos.
I will also do the unique_ptr publish method signature change in another PR, opening PRs in all the needed repos.

Complete CI:

• Linux
• Linux-aarch64
• macOS
• Windows

[wjwwood]

w.r.t. the splitting of the publisher and subscription headers, that's ok, but please also do the cpp files.

[ivanpauno]

One CI more:

• Linux
• Linux-aarch64
• macOS
• Windows

[ivanpauno]

CI:

• Linux
• Linux-aarch64
• macOS
• Windows

[ivanpauno]

Linux CI failed randomly, again:

• Linux