Reused shared memory with `PREALLOCATED_WITH_REALLOC`

[mauropasse]

On the PR ros2/rclcpp#2624 I provide a fix to allow the user to make copies of loaned messages, keeping them beyond the subscription callback scope. With the fix, the loaned message is returned to the DDS when the user's copy of the message goes out of scope.

In the PR I add a unit test to verify that memory is not reused until all entities have returned the loan.
The tests fully pass when history memory policy is set as DYNAMIC_REUSABLE, but fails when set as PREALLOCATED_WITH_REALLOC:

[  FAILED  ] 2 tests, listed below:
[  FAILED  ] TestSharedMemory.user_is_last_loan_owner
[  FAILED  ] TestSharedMemory.sub_is_last_loan_owner

The failures mean that a user is provided with loaned memory which has been previously provided, and is still in use:

• Case 1: The user keeps copies (loaned not returned)
• Case 2: A not-spinning subscription is keeping copies in its internal buffer.

While a temporary solution would be to only use DYNAMIC_REUSABLE on shared memory, this mode has the issue that a undesired copy is performed on a multi-process system, as I describe in #782

So in short:

PREALLOCATED_WITH_REALLOC: Issues shared memory still in use
DYNAMIC_REUSABLE: A copy of the message happens on a multi-process system

Do you think this is a bug, or the use of PREALLOCATED_WITH_REALLOC inherently breaks proper shared memory management?

[mauropasse]

These failures can be explained by the documentation in eProsima's datasharing-delivery-constraints:

Using Data-sharing mechanism, the DataWriter’s history is shared with the DataReaders.
This means that the effective HistoryQos depth on the DataReader is, at most, 
the Datawriter’s HistoryQos depth.

To avoid confusions, set the DataReaders’ history depth to a value equal or less than the DataWriter’s.

In the failure [ FAILED ] TestSharedMemory.sub_is_last_loan_owner, the subscription has set a QoS depth larger than the publisher depth, so the failure seems kind of expected.

My question is, should the subscription/reader log some warning in this case? Like some QoS mismatch event:
[WARNING]: HistoryQos depth on the DataReader should be equal or lesser than the DataWriter.

The goal would be for the user for somehow to know that if he wants shared memory to safely handle memory (i.e. its stored messages won't silently be overridden):

• Subscription depth should be <= Publisher depth
• User N copies should be <= Publisher depth

Maybe this makes sense to just be mentioned on the ROS2 documentation about shared memory & loaned messages.

[mauropasse]

Currently the user can still ask for Loaned Messages even if all the shared memory has been exhausted, like when the previous loaned messages hasn't been returned (either because the user has copies, or subscriptions has them in its buffers).

In this case, the user is provided with memory still in use, which will silently override user/subscription stored messages.

I think the right behavior would be for the publisher to throw when the user requests loans and memory has been exhausted. What do you think?

[fujitatomoya]

My question is, should the subscription/reader log some warning in this case? Like some QoS mismatch event:
[WARNING]: HistoryQos depth on the DataReader should be equal or lesser than the DataWriter.

IMO, sounds reasonable if they are doing data-sharing. and this is probably request for Fast-DDS, and we (ROS 2) can catch the QoS incompatibility event from rmw implementation?

In this case, the user is provided with memory still in use, which will silently override user/subscription stored messages.
I think the right behavior would be for the publisher to throw when the user requests loans and memory has been exhausted.

IMO this depends on the application requirement.

• If publisher stops publishing because the memory blocks are not returned by user, all the subscription waiting for the next messages and sub-system will stop immediately. if the 3rd party application put a hold on the shared memory pointer, they can stop the entire system. for me this is more like security issue. because of this, i am not sure stopping publishing would be the right way to take, including throwing exception.
• maybe publisher goes on publishing the data to overwrite the data? but there is a possibility that subscriber reads the corrupted data or overwritten one updated already by publisher. i think there has been a discussion on is_valid API to verify if the data is still intact on subscriber side, but not sure if this comes from DDS or vendor specific API. (I guess that is not the part of DDS, but vendor API can check SampleIfo.valid_data flag?) i think that this is_valid API would be useful for this case.
• Give some time to wait until the shared memory is released in the configuration. like https://fast-dds.docs.eprosima.com/en/latest/fastdds/transport/datasharing.html#blocking-reuse-of-samples-until-acknowledged
• after all, most safest way is currently implemented that disabling loaning the memory on subscriptions.

CC: @MiguelCompany @wjwwood @clalancette @alsora

[MiguelCompany]

@mauropasse Currently, the only way to make the publisher block till the shared payload is not being used is using RELIABLE, and KEEP_ALL.

You can also increase extra_samples in ResourceLimitsQosPolicy to make the pool bigger (for instance, having extra_samples = depth would make the pool twice the depth).

One thing that could be done is adding a warning when a sample is marked for reuse without it being acknowledged. There is this callback in the DataWriterListener that could be of use.

My question is, should the subscription/reader log some warning in this case? Like some QoS mismatch event:
[WARNING]: HistoryQos depth on the DataReader should be equal or lesser than the DataWriter.

The problem with this is that HistoryQos is not transmitted in discovery.

[mauropasse]

Great information @MiguelCompany. I'll explore the options!

[mauropasse]

@MiguelCompany I tested these scenarios based on your suggestions:

• KEEP_ALL on publisher: Shared memory isn't reused, even if loans were returned by the sub/user (since the publisher still keeps the loans). No memory corruption or duplicates detected, so it serves the purpose (but might not be the most performant option due possible dynamic allocations?)
• extra_samples: Works well; it extends the shared memory pool beyond the pub QoS depth. This option lets subs/users hold additional samples.

One thing that could be done is adding a warning when a sample is marked for reuse without it being acknowledged.

• I tested RELIABLE with on_unacknowledged_sample_removed: It didn't trigger, as the DDS sub likely sent the ACK back to pub even when loans weren’t returned? (subscription not spinning, storing messages). I overrided the API on custom_publisher_info.hpp: CustomDataWriterListener adding a warning, but it didn't show anything.

If the subscription is not spinning but it receives the loaned messages, I guess the ACK is still sent in DDS? So the on_unacknowledged_sample_removed wouldn't be useful in this case?
Thanks!

[MiguelCompany]

• KEEP_ALL on publisher: Shared memory isn't reused, even if loans were returned by the sub/user (since the publisher still keeps the loans). No memory corruption or duplicates detected, so it serves the purpose (but might not be the most performant option due possible dynamic allocations?)

No dynamic allocations here. The only possible issue is that, since the publisher is blocked until the oldest sample is acknowledged (or reliability.max_blocking_time elapses), it could hinder the publication rate.

• I tested RELIABLE with on_unacknowledged_sample_removed: It didn't trigger, as the DDS sub likely sent the ACK back to pub even when loans weren’t returned? (subscription not spinning, storing messages). I overrided the API on custom_publisher_info.hpp: CustomDataWriterListener adding a warning, but it didn't show anything.

That is awkward. Was the reader also RELIABLE? Samples are automatically acknowledged for BEST_EFFORT readers.
Just in case, could you check again with the changes on this commit? I just added an error log when the writer detects the situation of a sample being removed without acknowledgement.

[mauropasse]

That is awkward. Was the reader also RELIABLE?

Yes. So this is the scenario just to clarify:

Single process: [Pub / Sub_A / Sub_B] = RELIABLE
1. Publisher (depth=1) publishes loaned messages
2. Subscription_A  (depth=10) doesn't spin, stores messages
3. Subscription_B spin, user stores 10 messages

After publishing the 2nd message, the publisher is provided with same memory 
used in the 1st message, despite loans not being returned. This is where we'd like to see the warning. 

When is usually the ACK sent by the subscription? When executed or when message is received in DataReader?
What is the relation with this callback, and the fact that asking a loan will provide memory still in use?

The only case in which I manage to see this callback in action (with also the logs you added in the commit) is setting:

Pub: best effort
Sub: reliable

[WARN] [1728658509.233333338] [pub_sub_node]: New publisher discovered on topic '/topic',
   offering incompatible QoS. No messages will be sent to it. Last incompatible policy: RELIABILITY_QOS_POLICY

[INFO] [1728658509.243104727] [pub_sub_node]: Publishing: '1'
2024-10-11 14:55:09.243 [RTPS_HISTORY Error] Unacknowledged change removed: test_msgs::msg::dds_::BasicTypes_ -> Function prepare_change
[WARN] [1728658509.243487964] [rmw_fastrtps_shared_cpp]: CustomPublisherInfo: on_unacknowledged_sample_removed!!
memory reused: 0x56141ada4bc0

[INFO] [1728658509.253650552] [pub_sub_node]: Publishing: '2'
[WARN] [1728658509.253678907] [rmw_fastrtps_shared_cpp]: CustomPublisherInfo: on_unacknowledged_sample_removed!!
2024-10-11 14:55:09.253 [RTPS_HISTORY Error] Unacknowledged change removed: test_msgs::msg::dds_::BasicTypes_ -> Function prepare_change
memory reused: 0x56141ada4b50

[mauropasse]

Since my test is single process, the send_datasharing_ack() is not called, but I'm unsure if is the same ACK sender that causes the callback on_unacknowledged_sample_removed() to not be called.

[mauropasse]

I expanded the test to check multi-process behavior (pub/sub in different processes):

• Subscription not spinning (thus not returning loans):
  • The on_unacknowledged_sample_removed() callback is invoked correctly when memory is reused (since no ACK is sent to the publisher).
  • The user (publisher process) could be warned about memory still in use.
• Subscription spinning (but user keeps messages, not returning loans):
  • The callback is not invoked, even if memory is reused, likely because the subscription sent ACK to the publisher.
• Single-process:
  • The callback is not invoked, regardless of whether the subscription is spinning.

So, in most cases, the on_unacknowledged_sample_removed() callback doesn't serve to notify the user about memory reuse.

A different approach to verify if memory is still in use could be:

• If the publisher knows how many same-host subscriptions exist, it could infer if memory is still in use:
  • The loan must be returned X times (where X = N subscriptions + 1 for the transient_local publisher).

Could logic be implemented to track this? What do you think @MiguelCompany

[fujitatomoya]

@mauropasse thank you for checking this, for us this is really interesting topic!

Subscription spinning (but user keeps messages, not returning loans):
The callback is not invoked, even if memory is reused, likely because the subscription sent ACK to the publisher.

this is because that ACK is sent to the corresponding publisher when the message is taken on the subscription? right?

rmw_fastrtps/rmw_fastrtps_shared_cpp/src/rmw_take.cpp

Line 98 in 5cb7665

while (ReturnCode_t::RETCODE_OK == info->data_reader_->take(data_values, info_seq, 1)) {

for me, this looks like okay by design of on_unacknowledged_sample_removed? because subscription sent the ACK for that message.
i mean that on_unacknowledged_sample_removed does not tell the data is valid or not...

i think we need the is_valid API in the end, so that application can check the data integrity right before accessing the data...

Just FYI, with previous comment on #783 (comment)

Pub: best effort
Sub: reliable

I do not think this is gonna communicate at all...

offering incompatible QoS. No messages will be sent to it. Last incompatible policy: RELIABILITY_QOS_POLICY

DDS detected the incompatible QoS, and no messages are delivered to this concerned subscription. because DDS cannot guarantee RELIABLE durability on this subscription.

[MiguelCompany]

@mauropasse @fujitatomoya This is in fact getting interesting...

this is because that ACK is sent to the corresponding publisher when the message is taken on the subscription? right?

@fujitatomoya you nailed it! It seems Fast DDS is acknowledging the sample when taking it.

So the following table is a summary on when samples are acknowledged

	Same process	Inter-process without data-sharing	Inter-process with data-sharing
Sample acknowledged...	... when published	... when added to history	... on take

A different approach to verify if memory is still in use could be:

• If the publisher knows how many same-host subscriptions exist, it could infer if memory is still in use:

  • The loan must be returned X times (where X = N subscriptions + 1 for the transient_local publisher).

Could logic be implemented to track this? What do you think @MiguelCompany

I think it would be better to extend the current mechanism and make it possible to acknowledge the sample when the loan is returned. This way we could use on_unacknowledged_sample_removed to warn the user.

Either that or expose is_valid API as @fujitatomoya suggested

i think we need the is_valid API in the end, so that application can check the data integrity right before accessing the data...

[MiguelCompany]

I thought of another possibility. The user could be warned when an item is added to the LoanManager, and the address of the sample already exists there:

rmw_fastrtps/rmw_fastrtps_shared_cpp/src/rmw_take.cpp

Lines 534 to 539 in 9d2150f

	void add_item(
	std::unique_ptr<Item> item)
	{
	std::lock_guard<std::mutex> guard(mtx);
	items.push_back(std::move(item));
	}