diff --git a/.gitignore b/.gitignore new file mode 100644 index 00000000..722d5e71 --- /dev/null +++ b/.gitignore @@ -0,0 +1 @@ +.vscode diff --git a/rmw_zenoh_cpp/CMakeLists.txt b/rmw_zenoh_cpp/CMakeLists.txt index 69939fd9..7d8db9dd 100644 --- a/rmw_zenoh_cpp/CMakeLists.txt +++ b/rmw_zenoh_cpp/CMakeLists.txt @@ -11,6 +11,8 @@ if(CMAKE_COMPILER_IS_GNUCXX OR CMAKE_CXX_COMPILER_ID MATCHES "Clang") add_compile_options(-Wall -Wextra -Wpedantic) endif() +set(RMW_ZENOH_BUILD_WITH_SHARED_MEMORY ON CACHE BOOL "Compile Zenoh RMW with Shared Memory support") + # find dependencies find_package(ament_cmake REQUIRED) @@ -80,6 +82,13 @@ target_compile_definitions(rmw_zenoh_cpp RMW_VERSION_PATCH=${rmw_VERSION_PATCH} ) +if(${RMW_ZENOH_BUILD_WITH_SHARED_MEMORY}) + target_compile_definitions(rmw_zenoh_cpp + PRIVATE + RMW_ZENOH_BUILD_WITH_SHARED_MEMORY + ) +endif() + ament_export_targets(export_rmw_zenoh_cpp) register_rmw_implementation( diff --git a/rmw_zenoh_cpp/config/DEFAULT_RMW_ZENOH_ROUTER_CONFIG.json5 b/rmw_zenoh_cpp/config/DEFAULT_RMW_ZENOH_ROUTER_CONFIG.json5 index f55c8434..25e065dc 100644 --- a/rmw_zenoh_cpp/config/DEFAULT_RMW_ZENOH_ROUTER_CONFIG.json5 +++ b/rmw_zenoh_cpp/config/DEFAULT_RMW_ZENOH_ROUTER_CONFIG.json5 @@ -457,8 +457,6 @@ /// A probing procedure for shared memory is performed upon session opening. To enable zenoh to operate /// over shared memory (and to not fallback on network mode), shared memory needs to be enabled also on the /// subscriber side. By doing so, the probing procedure will succeed and shared memory will operate as expected. - /// - /// ROS setting: disabled by default until fully tested enabled: false, }, auth: { diff --git a/rmw_zenoh_cpp/config/DEFAULT_RMW_ZENOH_SESSION_CONFIG.json5 b/rmw_zenoh_cpp/config/DEFAULT_RMW_ZENOH_SESSION_CONFIG.json5 index daffd790..eaa570af 100644 --- a/rmw_zenoh_cpp/config/DEFAULT_RMW_ZENOH_SESSION_CONFIG.json5 +++ b/rmw_zenoh_cpp/config/DEFAULT_RMW_ZENOH_SESSION_CONFIG.json5 @@ -462,8 +462,6 @@ /// A probing procedure for shared memory is performed upon session opening. To enable zenoh to operate /// over shared memory (and to not fallback on network mode), shared memory needs to be enabled also on the /// subscriber side. By doing so, the probing procedure will succeed and shared memory will operate as expected. - /// - /// ROS setting: disabled by default until fully tested enabled: false, }, auth: { diff --git a/rmw_zenoh_cpp/src/detail/rmw_context_impl_s.cpp b/rmw_zenoh_cpp/src/detail/rmw_context_impl_s.cpp index 8e549394..4b77e031 100644 --- a/rmw_zenoh_cpp/src/detail/rmw_context_impl_s.cpp +++ b/rmw_zenoh_cpp/src/detail/rmw_context_impl_s.cpp @@ -90,19 +90,24 @@ rmw_context_impl_s::Data::Data( std::size_t domain_id, const std::string & enclave, z_owned_session_t session, - std::optional shm_provider, const std::string & liveliness_str, - std::shared_ptr graph_cache) + std::shared_ptr graph_cache +#ifdef RMW_ZENOH_BUILD_WITH_SHARED_MEMORY + , std::optional shm +#endif +) : enclave_(std::move(enclave)), domain_id_(std::move(domain_id)), session_(std::move(session)), - shm_provider_(std::move(shm_provider)), liveliness_str_(std::move(liveliness_str)), graph_cache_(std::move(graph_cache)), is_shutdown_(false), next_entity_id_(0), is_initialized_(false), nodes_({}) +#ifdef RMW_ZENOH_BUILD_WITH_SHARED_MEMORY + , shm_(shm) +#endif { graph_guard_condition_ = std::make_unique(); graph_guard_condition_->implementation_identifier = rmw_zenoh_cpp::rmw_zenoh_identifier; @@ -178,14 +183,18 @@ rmw_ret_t rmw_context_impl_s::Data::shutdown() } z_undeclare_subscriber(z_move(graph_subscriber_)); - if (shm_provider_.has_value()) { - z_drop(z_move(shm_provider_.value())); - } + // Don't touch Zenoh Session if the ROS process is exiting, // it will cause panic. if (!is_exiting) { z_close(z_loan_mut(session_), NULL); } + +#ifdef RMW_ZENOH_BUILD_WITH_SHARED_MEMORY + // drop SHM subsystem if used + shm_ = std::nullopt; +#endif + is_shutdown_ = true; return RMW_RET_OK; } @@ -214,14 +223,6 @@ rmw_context_impl_s::rmw_context_impl_s( throw std::runtime_error("Error configuring Zenoh session."); } - // Check if shm is enabled. - z_owned_string_t shm_enabled; - zc_config_get_from_str(z_loan(config), Z_CONFIG_SHARED_MEMORY_KEY, &shm_enabled); - auto always_free_shm_enabled = rcpputils::make_scope_exit( - [&shm_enabled]() { - z_drop(z_move(shm_enabled)); - }); - // Initialize the zenoh session. z_owned_session_t session; if (z_open(&session, z_move(config), NULL) != Z_OK) { @@ -309,40 +310,51 @@ rmw_context_impl_s::rmw_context_impl_s( } z_drop(z_move(handler)); - // Initialize the shm manager if shared_memory is enabled in the config. - std::optional shm_provider = std::nullopt; - if (strncmp(z_string_data(z_loan(shm_enabled)), "true", z_string_len(z_loan(shm_enabled))) == 0) { +#ifdef RMW_ZENOH_BUILD_WITH_SHARED_MEMORY + // Initialize the shm subsystem if shared_memory is enabled in the config + std::optional shm; + if (rmw_zenoh_cpp::zenoh_shm_enabled()) { RMW_ZENOH_LOG_DEBUG_NAMED("rmw_zenoh_cpp", "SHM is enabled"); - // TODO(yuyuan): determine the default alignment of SHM - z_alloc_alignment_t alignment = {5}; - z_owned_memory_layout_t layout; - z_memory_layout_new(&layout, SHM_BUFFER_SIZE_MB * 1024 * 1024, alignment); + rmw_zenoh_cpp::ShmContext shm_context; + + // Read msg size treshold from config + shm_context.msgsize_threshold = rmw_zenoh_cpp::zenoh_shm_message_size_threshold(); - z_owned_shm_provider_t provider; - if (z_posix_shm_provider_new(&provider, z_loan(layout)) != Z_OK) { - RMW_ZENOH_LOG_ERROR_NAMED("rmw_zenoh_cpp", "Unable to create a SHM provider."); - throw std::runtime_error("Unable to create shm manager."); + // Create Layout for provider's memory + // Provider's alignment will be 1 byte as we are going to make only 1-byte aligned allocations + // TODO(yellowhatter): use zenoh_shm_message_size_threshold as base for alignment + z_alloc_alignment_t alignment = {0}; + z_owned_memory_layout_t layout; + if (z_memory_layout_new(&layout, rmw_zenoh_cpp::zenoh_shm_alloc_size(), alignment) != Z_OK) { + throw std::runtime_error("Unable to create a Layout for SHM provider."); } - shm_provider = provider; + // Create SHM provider + const auto provider_creation_result = + z_posix_shm_provider_new(&shm_context.shm_provider, z_loan(layout)); + z_drop(z_move(layout)); + if (provider_creation_result != Z_OK) { + throw std::runtime_error("Unable to create an SHM provider."); + } + // Upon successful provider creation, store it in the context + shm = std::make_optional(std::move(shm_context)); + } else { + RMW_ZENOH_LOG_DEBUG_NAMED("rmw_zenoh_cpp", "SHM is disabled"); } - auto free_shm_provider = rcpputils::make_scope_exit( - [&shm_provider]() { - if (shm_provider.has_value()) { - z_drop(z_move(shm_provider.value())); - } - }); +#endif close_session.cancel(); - free_shm_provider.cancel(); data_ = std::make_shared( domain_id, std::move(enclave), std::move(session), - std::move(shm_provider), std::move(liveliness_str), - std::move(graph_cache)); + std::move(graph_cache) +#ifdef RMW_ZENOH_BUILD_WITH_SHARED_MEMORY + , std::move(shm) +#endif + ); ret = data_->subscribe_to_ros_graph(); if (ret != RMW_RET_OK) { @@ -376,11 +388,13 @@ const z_loaned_session_t * rmw_context_impl_s::session() const } ///============================================================================= -std::optional & rmw_context_impl_s::shm_provider() +#ifdef RMW_ZENOH_BUILD_WITH_SHARED_MEMORY +std::optional & rmw_context_impl_s::shm() { std::lock_guard lock(data_->mutex_); - return data_->shm_provider_; + return data_->shm_; } +#endif ///============================================================================= rmw_guard_condition_t * rmw_context_impl_s::graph_guard_condition() diff --git a/rmw_zenoh_cpp/src/detail/rmw_context_impl_s.hpp b/rmw_zenoh_cpp/src/detail/rmw_context_impl_s.hpp index c68aa44c..c99f5018 100644 --- a/rmw_zenoh_cpp/src/detail/rmw_context_impl_s.hpp +++ b/rmw_zenoh_cpp/src/detail/rmw_context_impl_s.hpp @@ -50,11 +50,13 @@ class rmw_context_impl_s final // create other Zenoh objects. const z_loaned_session_t * session() const; - // Get a reference to the shm_provider. +#ifdef RMW_ZENOH_BUILD_WITH_SHARED_MEMORY + // Get a reference to the shm subsystem. // Note: This is not thread-safe. // TODO(Yadunund): Remove this API and instead include a publish() API // that handles the shm_provider once the context manages publishers. - std::optional & shm_provider(); + std::optional & shm(); +#endif // Get the graph guard condition. rmw_guard_condition_t * graph_guard_condition(); @@ -103,9 +105,12 @@ class rmw_context_impl_s final std::size_t domain_id, const std::string & enclave, z_owned_session_t session, - std::optional shm_provider, const std::string & liveliness_str, - std::shared_ptr graph_cache); + std::shared_ptr graph_cache +#ifdef RMW_ZENOH_BUILD_WITH_SHARED_MEMORY + , std::optional shm +#endif + ); // Subscribe to the ROS graph. rmw_ret_t subscribe_to_ros_graph(); @@ -126,9 +131,6 @@ class rmw_context_impl_s final std::size_t domain_id_; // An owned session. z_owned_session_t session_; - // An optional SHM manager that is initialized of SHM is enabled in the - // zenoh session config. - std::optional shm_provider_; // Liveliness keyexpr string to subscribe to for ROS graph changes. std::string liveliness_str_; // Graph cache. @@ -148,6 +150,11 @@ class rmw_context_impl_s final bool is_initialized_; // Nodes created from this context. std::unordered_map> nodes_; +#ifdef RMW_ZENOH_BUILD_WITH_SHARED_MEMORY + // An optional SHM context that is initialized if SHM is enabled in the + // zenoh session config. + std::optional shm_; +#endif }; std::shared_ptr data_{nullptr}; diff --git a/rmw_zenoh_cpp/src/detail/rmw_publisher_data.cpp b/rmw_zenoh_cpp/src/detail/rmw_publisher_data.cpp index a3b78a3d..dfac42f4 100644 --- a/rmw_zenoh_cpp/src/detail/rmw_publisher_data.cpp +++ b/rmw_zenoh_cpp/src/detail/rmw_publisher_data.cpp @@ -223,8 +223,11 @@ PublisherData::PublisherData( ///============================================================================= rmw_ret_t PublisherData::publish( - const void * ros_message, - std::optional & shm_provider) + const void * ros_message +#ifdef RMW_ZENOH_BUILD_WITH_SHARED_MEMORY + , std::optional & shm +#endif +) { std::lock_guard lock(mutex_); if (is_shutdown_) { @@ -233,12 +236,14 @@ rmw_ret_t PublisherData::publish( } // Serialize data. - size_t max_data_length = type_support_->get_estimated_serialized_size( + const size_t max_data_length = type_support_->get_estimated_serialized_size( ros_message, type_support_impl_); // To store serialized message byte array. char * msg_bytes = nullptr; + +#ifdef RMW_ZENOH_BUILD_WITH_SHARED_MEMORY std::optional shmbuf = std::nullopt; auto always_free_shmbuf = rcpputils::make_scope_exit( [&shmbuf]() { @@ -246,24 +251,36 @@ rmw_ret_t PublisherData::publish( z_drop(z_move(shmbuf.value())); } }); +#endif rcutils_allocator_t * allocator = &rmw_node_->context->options.allocator; auto always_free_msg_bytes = rcpputils::make_scope_exit( - [&msg_bytes, allocator, &shmbuf]() { - if (msg_bytes && !shmbuf.has_value()) { + [&msg_bytes, allocator +#ifdef RMW_ZENOH_BUILD_WITH_SHARED_MEMORY + , &shmbuf +#endif + ]() { + if (msg_bytes +#ifdef RMW_ZENOH_BUILD_WITH_SHARED_MEMORY + && !shmbuf.has_value() +#endif + ) + { allocator->deallocate(msg_bytes, allocator->state); } }); +#ifdef RMW_ZENOH_BUILD_WITH_SHARED_MEMORY // Get memory from SHM buffer if available. - if (shm_provider.has_value()) { + if (shm.has_value() && max_data_length >= shm.value().msgsize_threshold) { RMW_ZENOH_LOG_DEBUG_NAMED("rmw_zenoh_cpp", "SHM is enabled."); - auto provider = shm_provider.value(); + auto & provider = shm.value().shm_provider; + + // TODO(yellowhatter): SHM, use alignment based on msgsize_threshold + z_alloc_alignment_t alignment = {0}; z_buf_layout_alloc_result_t alloc; - // TODO(yuyuan): SHM, configure this - z_alloc_alignment_t alignment = {5}; z_shm_provider_alloc_gc_defrag_blocking(&alloc, z_loan(provider), SHM_BUF_OK_SIZE, alignment); if (alloc.status == ZC_BUF_LAYOUT_ALLOC_STATUS_OK) { @@ -275,11 +292,14 @@ rmw_ret_t PublisherData::publish( return RMW_RET_ERROR; } } else { - // Get memory from the allocator. - msg_bytes = static_cast(allocator->allocate(max_data_length, allocator->state)); - RMW_CHECK_FOR_NULL_WITH_MSG( - msg_bytes, "bytes for message is null", return RMW_RET_BAD_ALLOC); - } +#endif + // Get memory from the allocator. + msg_bytes = static_cast(allocator->allocate(max_data_length, allocator->state)); + RMW_CHECK_FOR_NULL_WITH_MSG( + msg_bytes, "bytes for message is null", return RMW_RET_BAD_ALLOC); +#ifdef RMW_ZENOH_BUILD_WITH_SHARED_MEMORY +} +#endif // Object that manages the raw buffer eprosima::fastcdr::FastBuffer fastbuffer(msg_bytes, max_data_length); @@ -309,11 +329,15 @@ rmw_ret_t PublisherData::publish( options.attachment = z_move(attachment); z_owned_bytes_t payload; +#ifdef RMW_ZENOH_BUILD_WITH_SHARED_MEMORY if (shmbuf.has_value()) { z_bytes_from_shm_mut(&payload, z_move(shmbuf.value())); } else { - z_bytes_copy_from_buf(&payload, reinterpret_cast(msg_bytes), data_length); - } +#endif + z_bytes_copy_from_buf(&payload, reinterpret_cast(msg_bytes), data_length); +#ifdef RMW_ZENOH_BUILD_WITH_SHARED_MEMORY +} +#endif z_result_t res = z_publisher_put(z_loan(pub_), z_move(payload), &options); if (res != Z_OK) { @@ -332,8 +356,11 @@ rmw_ret_t PublisherData::publish( ///============================================================================= rmw_ret_t PublisherData::publish_serialized_message( - const rmw_serialized_message_t * serialized_message, - std::optional & /*shm_provider*/) + const rmw_serialized_message_t * serialized_message +#ifdef RMW_ZENOH_BUILD_WITH_SHARED_MEMORY + , std::optional & /*shm_provider*/ +#endif +) { eprosima::fastcdr::FastBuffer buffer( reinterpret_cast(serialized_message->buffer), serialized_message->buffer_length); diff --git a/rmw_zenoh_cpp/src/detail/rmw_publisher_data.hpp b/rmw_zenoh_cpp/src/detail/rmw_publisher_data.hpp index 56767ca4..eca222b7 100644 --- a/rmw_zenoh_cpp/src/detail/rmw_publisher_data.hpp +++ b/rmw_zenoh_cpp/src/detail/rmw_publisher_data.hpp @@ -27,6 +27,7 @@ #include "event.hpp" #include "liveliness_utils.hpp" #include "message_type_support.hpp" +#include "shm_context.hpp" #include "rmw/ret_types.h" @@ -49,13 +50,19 @@ class PublisherData final // Publish a ROS message. rmw_ret_t publish( - const void * ros_message, - std::optional & shm_provider); + const void * ros_message +#ifdef RMW_ZENOH_BUILD_WITH_SHARED_MEMORY + , std::optional & shm +#endif + ); // Publish a serialized ROS message. rmw_ret_t publish_serialized_message( - const rmw_serialized_message_t * serialized_message, - std::optional & shm_provider); + const rmw_serialized_message_t * serialized_message +#ifdef RMW_ZENOH_BUILD_WITH_SHARED_MEMORY + , std::optional & shm +#endif + ); // Get a copy of the keyexpr_hash of this PublisherData's liveliness::Entity. std::size_t keyexpr_hash() const; diff --git a/rmw_zenoh_cpp/src/detail/rmw_subscription_data.hpp b/rmw_zenoh_cpp/src/detail/rmw_subscription_data.hpp index 9f9f8c02..4f864994 100644 --- a/rmw_zenoh_cpp/src/detail/rmw_subscription_data.hpp +++ b/rmw_zenoh_cpp/src/detail/rmw_subscription_data.hpp @@ -34,6 +34,7 @@ #include "message_type_support.hpp" #include "attachment_helpers.hpp" #include "type_support_common.hpp" +#include "shm_context.hpp" #include "rcutils/allocator.h" @@ -74,8 +75,11 @@ class SubscriptionData final : public std::enable_shared_from_this & shm_provider); + const void * ros_message +#ifdef RMW_ZENOH_BUILD_WITH_SHARED_MEMORY + , std::optional & shm +#endif + ); // Get a copy of the keyexpr_hash of this SubscriptionData's liveliness::Entity. std::size_t keyexpr_hash() const; diff --git a/rmw_zenoh_cpp/src/detail/shm_context.cpp b/rmw_zenoh_cpp/src/detail/shm_context.cpp new file mode 100644 index 00000000..8dcf8566 --- /dev/null +++ b/rmw_zenoh_cpp/src/detail/shm_context.cpp @@ -0,0 +1,26 @@ +// Copyright 2023 Open Source Robotics Foundation, Inc. +// +// 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 "shm_context.hpp" + +namespace rmw_zenoh_cpp +{ +///============================================================================= +#ifdef RMW_ZENOH_BUILD_WITH_SHARED_MEMORY +ShmContext::~ShmContext() +{ + z_drop(z_move(shm_provider)); +} +#endif +} // namespace rmw_zenoh_cpp diff --git a/rmw_zenoh_cpp/src/detail/shm_context.hpp b/rmw_zenoh_cpp/src/detail/shm_context.hpp new file mode 100644 index 00000000..619afa3e --- /dev/null +++ b/rmw_zenoh_cpp/src/detail/shm_context.hpp @@ -0,0 +1,33 @@ +// Copyright 2023 Open Source Robotics Foundation, Inc. +// +// 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 DETAIL__SHM_CONTEXT_HPP_ +#define DETAIL__SHM_CONTEXT_HPP_ + + +namespace rmw_zenoh_cpp +{ +///============================================================================= +#ifdef RMW_ZENOH_BUILD_WITH_SHARED_MEMORY +struct ShmContext +{ + z_owned_shm_provider_t shm_provider; + size_t msgsize_threshold; + + ~ShmContext(); +}; +#endif +} // namespace rmw_zenoh_cpp + +#endif // DETAIL__SHM_CONTEXT_HPP_ diff --git a/rmw_zenoh_cpp/src/detail/zenoh_config.cpp b/rmw_zenoh_cpp/src/detail/zenoh_config.cpp index 206e2a92..9429b145 100644 --- a/rmw_zenoh_cpp/src/detail/zenoh_config.cpp +++ b/rmw_zenoh_cpp/src/detail/zenoh_config.cpp @@ -42,6 +42,14 @@ static const std::unordered_map zenoh_router_check_attempts() // If unset, check indefinitely. return default_value; } + +#ifdef RMW_ZENOH_BUILD_WITH_SHARED_MEMORY +///============================================================================= +bool zenoh_shm_enabled() +{ + const char * envar_value; + + if (NULL != rcutils_get_env(zenoh_shm_enabled_envar, &envar_value)) { + RMW_ZENOH_LOG_ERROR_NAMED( + "rmw_zenoh_cpp", "Envar %s cannot be read. Report this bug.", + zenoh_shm_enabled_envar); + return zenoh_shm_enabled_default; + } + + if (strncmp(envar_value, "false", strlen(envar_value)) == 0) { + return false; + } + + if (strncmp(envar_value, "true", strlen(envar_value)) == 0) { + return true; + } + + return zenoh_shm_enabled_default; +} +///============================================================================= +size_t zenoh_shm_alloc_size() +{ + const char * envar_value; + + if (NULL != rcutils_get_env(zenoh_shm_alloc_size_envar, &envar_value)) { + RMW_ZENOH_LOG_ERROR_NAMED( + "rmw_zenoh_cpp", "Envar %s cannot be read. Report this bug.", + zenoh_shm_alloc_size_envar); + return zenoh_shm_alloc_size_default; + } + + // If the environment variable contains a value, handle it accordingly. + if (envar_value[0] != '\0') { + const auto read_value = std::strtoull(envar_value, nullptr, 10); + if (read_value > 0) { + if (read_value > std::numeric_limits::max()) { + RMW_ZENOH_LOG_ERROR_NAMED( + "rmw_zenoh_cpp", "Envar %s: value is too large!", + zenoh_shm_alloc_size_envar); + } else { + return read_value; + } + } + } + + return zenoh_shm_alloc_size_default; +} +///============================================================================= +size_t zenoh_shm_message_size_threshold() +{ + const char * envar_value; + + if (NULL != rcutils_get_env(zenoh_shm_message_size_threshold_envar, &envar_value)) { + RMW_ZENOH_LOG_ERROR_NAMED( + "rmw_zenoh_cpp", "Envar %s cannot be read. Report this bug.", + zenoh_shm_message_size_threshold_envar); + return zenoh_shm_message_size_threshold_default; + } + + // If the environment variable contains a value, handle it accordingly. + if (envar_value[0] != '\0') { + const auto read_value = std::strtoull(envar_value, nullptr, 10); + if (read_value > 0) { + if (read_value > std::numeric_limits::max()) { + RMW_ZENOH_LOG_ERROR_NAMED( + "rmw_zenoh_cpp", "Envar %s: value is too large!", + zenoh_shm_message_size_threshold_envar); + } else if ((read_value & (read_value - 1)) != 0) { // power of 2 check + RMW_ZENOH_LOG_ERROR_NAMED( + "rmw_zenoh_cpp", "Envar %s: value must be power of 2!", + zenoh_shm_message_size_threshold_envar); + + } else { + return read_value; + } + } + } + + return zenoh_shm_message_size_threshold_default; +} +#endif } // namespace rmw_zenoh_cpp diff --git a/rmw_zenoh_cpp/src/detail/zenoh_config.hpp b/rmw_zenoh_cpp/src/detail/zenoh_config.hpp index 357c7e89..57aa657d 100644 --- a/rmw_zenoh_cpp/src/detail/zenoh_config.hpp +++ b/rmw_zenoh_cpp/src/detail/zenoh_config.hpp @@ -57,6 +57,36 @@ rmw_ret_t get_z_config(const ConfigurableEntity & entity, z_owned_config_t * con /// @return The number of times to try connecting to a zenoh router and /// std::nullopt if establishing a connection to a router is not required. std::optional zenoh_router_check_attempts(); + +#ifdef RMW_ZENOH_BUILD_WITH_SHARED_MEMORY +///============================================================================= +/// Get the enabled state of shared memory subsystem +/// based on the environment variable ZENOH_SHM_ENABLED. +/// @details The behavior is as follows: +/// - If not set or not "false", the default value of "true" is returned. +/// - Else "false" is returned. +/// @return The amount of shared memory to be pre-allocated for Zenoh SHM operation +bool zenoh_shm_enabled(); + +///============================================================================= +/// Get the amount of shared memory to be pre-allocated for Zenoh SHM operation +/// based on the environment variable ZENOH_SHM_ALLOC_SIZE. +/// @details The behavior is as follows: +/// - If not set or <= 0, the default value of 1MB is returned. +/// - Else value of environemnt variable is returned. +/// @return The amount of shared memory to be pre-allocated for Zenoh SHM operation +size_t zenoh_shm_alloc_size(); + +///============================================================================= +/// Message size threshold for implicit SHM optimization based on the environment +/// variable ZENOH_SHM_MESSAGE_SIZE_THRESHOLD. +/// Messages smaller than this threshold will not be forwarded through Zenoh SHM +/// @details The behavior is as follows: +/// - If not set or <= 0, the default value of 2KB is returned. +/// - Else value of environemnt variable is returned. +/// @return The amount of shared memory to be pre-allocated for Zenoh SHM operation +size_t zenoh_shm_message_size_threshold(); +#endif } // namespace rmw_zenoh_cpp #endif // DETAIL__ZENOH_CONFIG_HPP_ diff --git a/rmw_zenoh_cpp/src/rmw_zenoh.cpp b/rmw_zenoh_cpp/src/rmw_zenoh.cpp index 920b12dd..7cf350c2 100644 --- a/rmw_zenoh_cpp/src/rmw_zenoh.cpp +++ b/rmw_zenoh_cpp/src/rmw_zenoh.cpp @@ -598,8 +598,11 @@ rmw_publish( } return pub_data->publish( - ros_message, - context_impl->shm_provider()); + ros_message +#ifdef RMW_ZENOH_BUILD_WITH_SHARED_MEMORY + , context_impl->shm() +#endif + ); } //============================================================================== @@ -705,8 +708,11 @@ rmw_publish_serialized_message( RMW_CHECK_ARGUMENT_FOR_NULL(publisher_data, RMW_RET_INVALID_ARGUMENT); return publisher_data->publish_serialized_message( - serialized_message, - context_impl->shm_provider()); + serialized_message +#ifdef RMW_ZENOH_BUILD_WITH_SHARED_MEMORY + , context_impl->shm() +#endif + ); } //============================================================================== diff --git a/zenoh_c_vendor/CMakeLists.txt b/zenoh_c_vendor/CMakeLists.txt index ded765de..54401dd4 100644 --- a/zenoh_c_vendor/CMakeLists.txt +++ b/zenoh_c_vendor/CMakeLists.txt @@ -15,7 +15,9 @@ find_package(ament_cmake_vendor_package REQUIRED) # Note: We separate the two args needed for cargo with "$" and not ";" as the # latter is a list separater in cmake and hence the string will be split into two # when expanded. -set(ZENOHC_CARGO_FLAGS "--no-default-features$--features=shared-memory zenoh/transport_compression zenoh/transport_tcp zenoh/transport_tls") +set(ZENOHC_CARGO_FLAGS "--no-default-features$--features=zenoh/transport_compression zenoh/transport_tcp zenoh/transport_tls") + +set(RMW_ZENOH_BUILD_WITH_SHARED_MEMORY ON CACHE BOOL "Compile Zenoh RMW with Shared Memory support") # Set VCS_VERSION to include latest changes from zenoh-c to benefit from : # - https://github.com/eclipse-zenoh/zenoh-c/pull/340 (fix build issue) @@ -29,6 +31,7 @@ ament_vendor(zenoh_c_vendor VCS_VERSION a57fda736d2f8866c06509f34b5bc73e372c6aaf CMAKE_ARGS "-DZENOHC_CARGO_FLAGS=${ZENOHC_CARGO_FLAGS}" + "-DZENOHC_BUILD_WITH_SHARED_MEMORY=${RMW_ZENOH_BUILD_WITH_SHARED_MEMORY}" "-DZENOHC_BUILD_WITH_UNSTABLE_API=TRUE" "-DZENOHC_CUSTOM_TARGET=${ZENOHC_CUSTOM_TARGET}" )