diff --git a/src/cpp/rtps/DataSharing/DataSharingNotification.hpp b/src/cpp/rtps/DataSharing/DataSharingNotification.hpp index e8b94ebd585..6307dd18747 100644 --- a/src/cpp/rtps/DataSharing/DataSharingNotification.hpp +++ b/src/cpp/rtps/DataSharing/DataSharingNotification.hpp @@ -134,17 +134,18 @@ class DataSharingNotification { segment_id_ = reader_guid; segment_name_ = generate_segment_name(shared_dir, reader_guid); - - uint32_t per_allocation_extra_size = T::compute_per_allocation_extra_size( - alignof(Notification), DataSharingNotification::domain_name()); - uint32_t segment_size = static_cast(sizeof(Notification)) + per_allocation_extra_size; - - //Open the segment - T::remove(segment_name_); std::unique_ptr local_segment; + try { - local_segment = std::unique_ptr( + uint32_t per_allocation_extra_size = T::compute_per_allocation_extra_size( + alignof(Notification), DataSharingNotification::domain_name()); + uint32_t segment_size = static_cast(sizeof(Notification)) + per_allocation_extra_size; + + //Open the segment + T::remove(segment_name_); + + local_segment.reset( new T(boost::interprocess::create_only, segment_name_, segment_size + T::EXTRA_SEGMENT_SIZE)); diff --git a/src/cpp/rtps/DataSharing/WriterPool.hpp b/src/cpp/rtps/DataSharing/WriterPool.hpp index 1a78c1242dd..5584a5e476e 100644 --- a/src/cpp/rtps/DataSharing/WriterPool.hpp +++ b/src/cpp/rtps/DataSharing/WriterPool.hpp @@ -147,45 +147,50 @@ class WriterPool : public DataSharingPayloadPool writer_ = writer; segment_id_ = writer_->getGuid(); segment_name_ = generate_segment_name(shared_dir, segment_id_); - - // We need to reserve the whole segment at once, and the underlying classes use uint32_t as size type. - // In order to avoid overflows, we will calculate using uint64 and check the casting - bool overflow = false; - size_t per_allocation_extra_size = T::compute_per_allocation_extra_size( - alignof(PayloadNode), DataSharingPayloadPool::domain_name()); - size_t payload_size = DataSharingPayloadPool::node_size(max_data_size_); - - uint64_t estimated_size_for_payloads_pool = pool_size_ * payload_size; - overflow |= (estimated_size_for_payloads_pool != static_cast(estimated_size_for_payloads_pool)); - uint32_t size_for_payloads_pool = static_cast(estimated_size_for_payloads_pool); - - //Reserve one extra to avoid pointer overlapping - uint64_t estimated_size_for_history = (pool_size_ + 1) * sizeof(Segment::Offset); - overflow |= (estimated_size_for_history != static_cast(estimated_size_for_history)); - uint32_t size_for_history = static_cast(estimated_size_for_history); - - uint32_t descriptor_size = static_cast(sizeof(PoolDescriptor)); - uint64_t estimated_segment_size = size_for_payloads_pool + per_allocation_extra_size + - size_for_history + per_allocation_extra_size + - descriptor_size + per_allocation_extra_size; - overflow |= (estimated_segment_size != static_cast(estimated_segment_size)); - uint32_t segment_size = static_cast(estimated_segment_size); - - if (overflow) - { - logError(DATASHARING_PAYLOADPOOL, "Failed to create segment " << segment_name_ - << ": Segment size is too large: " << estimated_size_for_payloads_pool - << " (max is " << std::numeric_limits::max() << ")." - << " Please reduce the maximum size of the history"); - return false; - } - - //Open the segment - T::remove(segment_name_); std::unique_ptr local_segment; + size_t payload_size; + uint64_t estimated_size_for_payloads_pool; + uint64_t estimated_size_for_history; + uint32_t size_for_payloads_pool; + try { - local_segment = std::unique_ptr( + // We need to reserve the whole segment at once, and the underlying classes use uint32_t as size type. + // In order to avoid overflows, we will calculate using uint64 and check the casting + bool overflow = false; + size_t per_allocation_extra_size = T::compute_per_allocation_extra_size( + alignof(PayloadNode), DataSharingPayloadPool::domain_name()); + payload_size = DataSharingPayloadPool::node_size(max_data_size_); + + estimated_size_for_payloads_pool = pool_size_ * payload_size; + overflow |= (estimated_size_for_payloads_pool != static_cast(estimated_size_for_payloads_pool)); + size_for_payloads_pool = static_cast(estimated_size_for_payloads_pool); + + //Reserve one extra to avoid pointer overlapping + estimated_size_for_history = (pool_size_ + 1) * sizeof(Segment::Offset); + overflow |= (estimated_size_for_history != static_cast(estimated_size_for_history)); + uint32_t size_for_history = static_cast(estimated_size_for_history); + + uint32_t descriptor_size = static_cast(sizeof(PoolDescriptor)); + uint64_t estimated_segment_size = size_for_payloads_pool + per_allocation_extra_size + + size_for_history + per_allocation_extra_size + + descriptor_size + per_allocation_extra_size; + overflow |= (estimated_segment_size != static_cast(estimated_segment_size)); + uint32_t segment_size = static_cast(estimated_segment_size); + + if (overflow) + { + logError(DATASHARING_PAYLOADPOOL, "Failed to create segment " << segment_name_ + << ": Segment size is too large: " << estimated_size_for_payloads_pool + << " (max is " << std::numeric_limits::max() << ")." + << " Please reduce the maximum size of the history"); + return false; + } + + //Open the segment + T::remove(segment_name_); + + local_segment.reset( new T(boost::interprocess::create_only, segment_name_, segment_size + T::EXTRA_SEGMENT_SIZE)); diff --git a/src/cpp/rtps/transport/shared_mem/SharedMemManager.hpp b/src/cpp/rtps/transport/shared_mem/SharedMemManager.hpp index d7e1c1723ab..a93ad82e7c0 100644 --- a/src/cpp/rtps/transport/shared_mem/SharedMemManager.hpp +++ b/src/cpp/rtps/transport/shared_mem/SharedMemManager.hpp @@ -251,9 +251,21 @@ class SharedMemManager : " characters"); } - uint32_t extra_size = - SharedMemSegment::compute_per_allocation_extra_size(std::alignment_of::value, domain_name); - return std::shared_ptr(new SharedMemManager(domain_name, extra_size)); + try + { + uint32_t extra_size = + SharedMemSegment::compute_per_allocation_extra_size(std::alignment_of::value, + domain_name); + return std::shared_ptr(new SharedMemManager(domain_name, extra_size)); + } + catch (const std::exception& e) + { + logError(RTPS_TRANSPORT_SHM, "Failed to create Shared Memory Manager for domain " << domain_name + << ": " << + e.what()); + return std::shared_ptr(); + } + } ~SharedMemManager() diff --git a/src/cpp/rtps/transport/shared_mem/SharedMemTransport.cpp b/src/cpp/rtps/transport/shared_mem/SharedMemTransport.cpp index 2590d6a8664..d33717dba35 100644 --- a/src/cpp/rtps/transport/shared_mem/SharedMemTransport.cpp +++ b/src/cpp/rtps/transport/shared_mem/SharedMemTransport.cpp @@ -254,6 +254,10 @@ bool SharedMemTransport::init( try { shared_mem_manager_ = SharedMemManager::create(SHM_MANAGER_DOMAIN); + if (!shared_mem_manager_) + { + return false; + } shared_mem_segment_ = shared_mem_manager_->create_segment(configuration_.segment_size(), configuration_.port_queue_capacity()); diff --git a/src/cpp/utils/shared_memory/SharedMemSegment.hpp b/src/cpp/utils/shared_memory/SharedMemSegment.hpp index b3cdf43c94d..ada77ca9adf 100644 --- a/src/cpp/utils/shared_memory/SharedMemSegment.hpp +++ b/src/cpp/utils/shared_memory/SharedMemSegment.hpp @@ -326,7 +326,7 @@ class SharedSegment : public SharedSegmentBase } /** - * Estimates the extra segment space required for an allocation + * Estimates the extra segment space required for an allocation. This may throw */ static uint32_t compute_per_allocation_extra_size( size_t allocation_alignment, diff --git a/test/mock/rtps/SharedMemSegment/utils/shared_memory/SharedMemSegment.hpp b/test/mock/rtps/SharedMemSegment/utils/shared_memory/SharedMemSegment.hpp new file mode 100644 index 00000000000..a195d352bcf --- /dev/null +++ b/test/mock/rtps/SharedMemSegment/utils/shared_memory/SharedMemSegment.hpp @@ -0,0 +1,419 @@ +// Copyright 2019 Proyectos y Sistemas de Mantenimiento SL (eProsima). +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +#ifndef _FASTDDS_SHAREDMEM_SEGMENT_H_ +#define _FASTDDS_SHAREDMEM_SEGMENT_H_ + +#include + +// For gcc-9 disable a new warning that warns about copy operator implicitly +// defined based on copy constructor. +#if defined(__GNUC__) && ( __GNUC__ >= 9) +#pragma GCC diagnostic error "-Wdeprecated-copy" +#pragma GCC diagnostic push +#pragma GCC diagnostic ignored "-Wdeprecated-copy" +#endif // if defined(__GNUC__) && ( __GNUC__ >= 9) + +#include +#include + +#if defined (__GNUC__) && ( __GNUC__ >= 9) +#pragma GCC diagnostic pop +#endif // if defined (__GNUC__) && ( __GNUC__ >= 9) + +#include +#include +#include +#include +#include +#include +#include + +#include +#include + +namespace eprosima { +namespace fastdds { +namespace rtps { + +using Log = fastdds::dds::Log; + +/** + * Provides shared memory functionallity abstrating from + * lower level layers + */ +class SharedSegmentBase +{ +public: + + template + using sharable_lock = boost::interprocess::sharable_lock; + using sharable_mutex = boost::interprocess::interprocess_sharable_mutex; + + using condition_variable = RobustInterprocessCondition; + using mutex = boost::interprocess::interprocess_mutex; + using named_mutex = boost::interprocess::named_mutex; + using spin_wait = boost::interprocess::spin_wait; + + // Offset must be the same size for 32/64-bit versions, so no size_t used here. + using Offset = std::uint32_t; + + static constexpr boost::interprocess::open_only_t open_only = boost::interprocess::open_only_t(); + static constexpr boost::interprocess::create_only_t create_only = boost::interprocess::create_only_t(); + static constexpr boost::interprocess::open_or_create_t open_or_create = boost::interprocess::open_or_create_t(); + + // Boost memory manager needs extra memory to maintain its structures, + // as these structures are shared they are stored in the segment. + // TODO(Adolfo): Further analysis to determine the perfect value for this extra segment size + static constexpr uint32_t EXTRA_SEGMENT_SIZE = 512; + + explicit SharedSegmentBase( + const std::string& name) + : name_(name) + { + } + + virtual ~SharedSegmentBase() + { + } + + virtual void remove() = 0; + + std::string name() + { + return name_; + } + + virtual void* get_address_from_offset( + SharedSegmentBase::Offset offset) const = 0; + + virtual SharedSegmentBase::Offset get_offset_from_address( + void* address) const = 0; + + static std::unique_ptr open_or_create_and_lock_named_mutex( + const std::string& mutex_name) + { + std::unique_ptr named_mutex; + + named_mutex = std::unique_ptr( + new SharedSegmentBase::named_mutex(boost::interprocess::open_or_create, mutex_name.c_str())); + + boost::posix_time::ptime wait_time + = boost::posix_time::microsec_clock::universal_time() + + boost::posix_time::milliseconds(BOOST_INTERPROCESS_TIMEOUT_WHEN_LOCKING_DURATION_MS * 2); + if (!named_mutex->timed_lock(wait_time)) + { + // Interprocess mutex timeout when locking. Possible deadlock: owner died without unlocking? + // try to remove and create again + SharedSegmentBase::named_mutex::remove(mutex_name.c_str()); + + named_mutex = std::unique_ptr( + new SharedSegmentBase::named_mutex(boost::interprocess::open_or_create, mutex_name.c_str())); + + if (!named_mutex->try_lock()) + { + throw std::runtime_error("Couldn't create name_mutex: " + mutex_name); + } + } + + return named_mutex; + } + + static std::unique_ptr try_open_and_lock_named_mutex( + const std::string& mutex_name) + { + std::unique_ptr named_mutex; + + named_mutex = std::unique_ptr( + new SharedSegmentBase::named_mutex(boost::interprocess::open_only, mutex_name.c_str())); + + boost::posix_time::ptime wait_time + = boost::posix_time::microsec_clock::universal_time() + + boost::posix_time::milliseconds(BOOST_INTERPROCESS_TIMEOUT_WHEN_LOCKING_DURATION_MS * 2); + if (!named_mutex->timed_lock(wait_time)) + { + throw std::runtime_error("Couldn't lock name_mutex: " + mutex_name); + } + + return named_mutex; + } + + static std::unique_ptr open_named_mutex( + const std::string& mutex_name) + { + std::unique_ptr named_mutex; + + // Todo(Adolfo) : Dataraces could occur, this algorithm has to be improved + + named_mutex = std::unique_ptr( + new SharedSegmentBase::named_mutex(boost::interprocess::open_only, mutex_name.c_str())); + + return named_mutex; + } + + /** + * Unique ID of the segment + */ + class Id + { + public: + + typedef UUID<8> type; + + Id() + : uuid_(type::null_t{}) + { + } + + Id( + const Id& other) + { + uuid_ = other.uuid_; + } + + Id( + const type& uuid) + { + uuid_ = uuid; + } + + void generate() + { + type::generate(uuid_); + } + + const type& get() const + { + return uuid_; + } + + Id& operator = ( + const Id& other) + { + uuid_ = other.uuid_; + return *this; + } + + bool operator == ( + const Id& other) const + { + return uuid_ == other.uuid_; + } + + std::string to_string() + { + return uuid_.to_string(); + } + + static const Id null() + { + return Id(type(type::null_t())); + } + + private: + + type uuid_; + + }; // Id + +private: + + class EnvironmentInitializer + { + public: + + EnvironmentInitializer() + { + SharedMemEnvironment::get().init(); + } + + } + shared_mem_environment_initializer_; + + std::string name_; +}; + +template +class SharedSegment : public SharedSegmentBase +{ +public: + + typedef T managed_shared_memory_type; + typedef U managed_shared_object_type; + + SharedSegment( + boost::interprocess::create_only_t, + const std::string& name, + size_t size) + : SharedSegmentBase(name) + { + segment_ = std::unique_ptr( + new managed_shared_memory_type(boost::interprocess::create_only, name.c_str(), + static_cast(size + EXTRA_SEGMENT_SIZE))); + } + + SharedSegment( + boost::interprocess::open_only_t, + const std::string& name) + : SharedSegmentBase(name) + { + segment_ = std::unique_ptr( + new managed_shared_memory_type(boost::interprocess::open_only, name.c_str())); + } + + SharedSegment( + boost::interprocess::open_read_only_t, + const std::string& name) + : SharedSegmentBase(name) + { + segment_ = std::unique_ptr( + new managed_shared_memory_type(boost::interprocess::open_read_only, name.c_str())); + } + + SharedSegment( + boost::interprocess::open_or_create_t, + const std::string& name, + size_t size) + : SharedSegmentBase(name) + { + segment_ = std::unique_ptr( + new managed_shared_memory_type(boost::interprocess::create_only, name.c_str(), static_cast(size))); + } + + ~SharedSegment() + { + // no need of exception handling cause never throws + segment_.reset(); + } + + static void remove( + const std::string& name) + { + managed_shared_object_type::remove(name.c_str()); + } + + void remove() override + { + managed_shared_object_type::remove(name().c_str()); + } + + void* get_address_from_offset( + SharedSegment::Offset offset) const override + { + return segment_->get_address_from_handle(offset); + } + + SharedSegment::Offset get_offset_from_address( + void* address) const override + { + return segment_->get_handle_from_address(address); + } + + managed_shared_memory_type& get() + { + return *segment_; + } + + /** + * Estimates the extra segment space required for an allocation. This may throw + */ + static uint32_t compute_per_allocation_extra_size( + size_t allocation_alignment, + const std::string& domain_name) + { + Id uuid; + + try + { + static uint32_t extra_size = 0; + + if (extra_size == 0) + { + uuid.generate(); + + // Additional invalid path characters to trigger the exception + auto name = "///" + domain_name + "_" + uuid.to_string(); + + SharedMemEnvironment::get().init(); + + { + managed_shared_memory_type + test_segment(boost::interprocess::create_only, name.c_str(), + (std::max)((uint32_t)1024, static_cast(allocation_alignment * 4))); + + auto m1 = test_segment.get_free_memory(); + test_segment.allocate_aligned(1, static_cast(allocation_alignment)); + auto m2 = test_segment.get_free_memory(); + extra_size = static_cast(m1 - m2); + } + + managed_shared_object_type::remove(name.c_str()); + } + + return extra_size; + + } + catch (const std::exception& e) + { + logError(RTPS_TRANSPORT_SHM, "Failed to create segment " << uuid.to_string() + << ": " << e.what()); + + throw; + } + } + + /** + * Check the allocator internal structures + * @return true if structures are ok, false otherwise + */ + bool check_sanity() + { + return segment_->check_sanity(); + } + + /** + * @return The segment's size in bytes, including internal structures overhead. + */ + Offset mem_size() const + { + return segment_->get_size(); + } + +private: + + std::unique_ptr segment_; +}; + +using SharedMemSegment = SharedSegment< + boost::interprocess::basic_managed_shared_memory< + char, + boost::interprocess::rbtree_best_fit>, + boost::interprocess::iset_index>, + boost::interprocess::shared_memory_object>; + +using SharedFileSegment = SharedSegment< + boost::interprocess::basic_managed_mapped_file< + char, + boost::interprocess::rbtree_best_fit>, + boost::interprocess::iset_index>, + boost::interprocess::file_mapping>; + +} // namespace rtps +} // namespace fastdds +} // namespace eprosima + +#endif // _FASTDDS_SHAREDMEM_SEGMENT_H_ + diff --git a/test/unittest/CMakeLists.txt b/test/unittest/CMakeLists.txt index 028bbb2098c..069676942b7 100644 --- a/test/unittest/CMakeLists.txt +++ b/test/unittest/CMakeLists.txt @@ -13,6 +13,7 @@ # limitations under the License. add_subdirectory(rtps/common) +add_subdirectory(rtps/DataSharing) add_subdirectory(rtps/builtin) add_subdirectory(rtps/reader) add_subdirectory(rtps/writer) diff --git a/test/unittest/rtps/DataSharing/CMakeLists.txt b/test/unittest/rtps/DataSharing/CMakeLists.txt new file mode 100644 index 00000000000..76a77dab74f --- /dev/null +++ b/test/unittest/rtps/DataSharing/CMakeLists.txt @@ -0,0 +1,50 @@ +# Copyright 2023 Proyectos y Sistemas de Mantenimiento SL (eProsima). +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +set(SHMSEGMENTTESTS_SOURCE SHMSegmentTests.cpp + ${PROJECT_SOURCE_DIR}/src/cpp/rtps/DataSharing/DataSharingPayloadPool.cpp + ) + +if(WIN32) + add_definitions(-D_WIN32_WINNT=0x0601) +endif() + +add_executable(SHMSegmentTests ${SHMSEGMENTTESTS_SOURCE}) +target_compile_definitions(SHMSegmentTests PRIVATE FASTRTPS_NO_LIB + BOOST_ASIO_STANDALONE + ASIO_STANDALONE + $<$>,$>:__DEBUG> + $<$:__INTERNALDEBUG> # Internal debug activated. + $<$:_ENABLE_ATOMIC_ALIGNMENT_FIX> + $<$:NOMINMAX> # avoid conflict with std::min & std::max in visual studio + ) +target_include_directories(SHMSegmentTests PRIVATE + ${Asio_INCLUDE_DIR} + ${PROJECT_SOURCE_DIR}/test/mock/rtps/SharedMemSegment + ${PROJECT_SOURCE_DIR}/include + ${PROJECT_BINARY_DIR}/include + ${PROJECT_SOURCE_DIR}/src/cpp + ${THIRDPARTY_BOOST_INCLUDE_DIR} + ) +target_link_libraries(SHMSegmentTests + fastcdr fastrtps foonathan_memory + GTest::gmock + ${CMAKE_DL_LIBS} + ${CMAKE_DL_LIBS} + ${THIRDPARTY_BOOST_LINK_LIBS}) +add_gtest(SHMSegmentTests SOURCES ${SHMSEGMENTTESTS_SOURCE}) + +if(ANDROID) + set_property(TARGET SHMSegmentTests PROPERTY CROSSCOMPILING_EMULATOR "adb;shell;cd;${CMAKE_CURRENT_BINARY_DIR};&&") +endif() diff --git a/test/unittest/rtps/DataSharing/SHMSegmentTests.cpp b/test/unittest/rtps/DataSharing/SHMSegmentTests.cpp new file mode 100644 index 00000000000..e513e7219e6 --- /dev/null +++ b/test/unittest/rtps/DataSharing/SHMSegmentTests.cpp @@ -0,0 +1,81 @@ +// Copyright 2023 Proyectos y Sistemas de Mantenimiento SL (eProsima). +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +#include "fastdds/rtps/common/Types.h" +#include "fastrtps/qos/QosPolicies.h" +#include +#include + +#include +#include +#include +#include + +#include +#include +#include + +#include + +namespace eprosima { +namespace fastrtps { +namespace rtps { + +using namespace testing; + +// This test will attempt to create enable DataSharing while being unable to compute the segment size due to a lack +// of permissions. This should fail but not propagate an uncaught boost::interprocess::interprocess_exception +TEST(SHMSegmentTests, Writer) +{ + RTPSParticipantAttributes p_attr; + p_attr.builtin.discovery_config.discoveryProtocol = eprosima::fastrtps::rtps::DiscoveryProtocol::SIMPLE; + p_attr.builtin.use_WriterLivelinessProtocol = true; + RTPSParticipant* participant = RTPSDomain::createParticipant( + 0, p_attr); + + ASSERT_NE(participant, nullptr); + + HistoryAttributes history_attr; + history_attr.payloadMaxSize = 255; + history_attr.maximumReservedCaches = 50; + WriterHistory* history = new WriterHistory(history_attr); + WriterAttributes writer_attr; + DataSharingQosPolicy dsq; + // We select a folder to force the use of SharedFileSegment + dsq.automatic("tmp"); + writer_attr.endpoint.set_data_sharing_configuration(dsq); + + std::shared_ptr payload_pool(new WriterPool(history_attr.payloadMaxSize, + history_attr.maximumReservedCaches)); + + RTPSWriter* writer; + EXPECT_NO_THROW(writer = RTPSDomain::createRTPSWriter(participant, writer_attr, payload_pool, history)); + // RTPSWriter creation failed, as expected. + EXPECT_EQ(writer, nullptr); + + RTPSDomain::removeRTPSParticipant(participant); + delete(history); +} + +} // namespace rtps +} // namespace fastrtps +} // namespace eprosima + +int main( + int argc, + char** argv) +{ + testing::InitGoogleMock(&argc, argv); + return RUN_ALL_TESTS(); +}