实现自定义内存分配器

0. 简介

本教程将教你如何为发布器和侦听器集成自定义分配器，以便在执行ROS节点时永远不会调用默认堆分配器。本教程的代码在这里（https://github.com/ros2/demos/blob/foxy/demo_nodes_cpp/src/topics/allocator_tutorial.cpp）。

1.背景

假设您想要编写实时安全的代码，并且你已经听说了在实时临界区调用“new”的许多危险，因为大多数平台上的默认堆分配器是不确定性的。

默认情况下，许多c++标准库结构会在增长时隐式分配内存，比如std::vector。然而，这些数据结构也接受一个“Allocator”模板参数。如果你为这些数据结构之一指定一个自定义分配器，它将为你使用该分配器而不是系统分配器来增长或收缩数据结构。你的自定义分配器可以在堆栈上预分配内存池，这可能更适合于实时应用程序。

在ROS2 C++客户端库(rclcpp)中，我们遵循与c++标准库类似的理念。发布器、侦听器和Executor接受一个Allocator模板参数，该参数控制该实体在执行期间进行的分配。

2.编写一个分配器

要编写一个与ROS2的分配器接口兼容的分配器，你的分配器必须与c++标准库分配器接口兼容。

c++ 11库提供了一个名为allocator_traits的东西。c++ 11标准规定，自定义分配器只需要满足以标准方式分配和释放内存所需的最小需求集。allocator_traits是一个通用结构，它基于用最小需求编写的分配器来填充分配器的其他特性。

例如，下面的自定义分配器声明将满足allocator_traits(当然，你仍然需要在这个结构中实现声明的函数):

template <class T>
struct custom_allocator {
    
  using value_type = T;
  custom_allocator() noexcept;
  template <class U> custom_allocator (const custom_allocator<U>&) noexcept;
  T* allocate (std::size_t n);
  void deallocate (T* p, std::size_t n);
};

template <class T, class U>
constexpr bool operator== (const custom_allocator<T>&, const custom_allocator<U>&) noexcept;

template <class T, class U>
constexpr bool operator!= (const custom_allocator<T>&, const custom_allocator<U>&) noexcept;

然后你可以像这样访问由allocator_traits填充的其他函数和分配器成员:

std::allocator_traits<custom_allocator<T>>::construct(...)

要了解allocator_traits的全部功能，请参见https://en.cppreference.com/w/cpp/memory/allocator_traits。

然而，一些只支持部分c++ 11的编译器，如GCC 4.8，仍然需要分配器来实现大量的样本代码，以处理标准库结构(如vector和string)，因为这些结构在内部不使用allocator_traits。因此，如果你正在使用一个部分支持c++ 11的编译器，你的分配器将需要看起来更像这样:

template<typename T>
struct pointer_traits {
    
  using reference = T &;
  using const_reference = const T &;
};

// Avoid declaring a reference to void with an empty specialization
template<>
struct pointer_traits<void> {
    
};

template<typename T = void>
struct MyAllocator : public pointer_traits<T> {
    
public:
  using value_type = T;
  using size_type = std::size_t;
  using pointer = T *;
  using const_pointer = const T *;
  using difference_type = typename std::pointer_traits<pointer>::difference_type;

  MyAllocator() noexcept;

  ~MyAllocator() noexcept;

  template<typename U>
  MyAllocator(const MyAllocator<U> &) noexcept;

  T * allocate(size_t size, const void * = 0);

  void deallocate(T * ptr, size_t size);

  template<typename U>
  struct rebind {
    
    typedef MyAllocator<U> other;
  };
};

template<typename T, typename U>
constexpr bool operator==(const MyAllocator<T> &,
  const MyAllocator<U> &) noexcept;

template<typename T, typename U>
constexpr bool operator!=(const MyAllocator<T> &,
  const MyAllocator<U> &) noexcept;

3.编写一个main示例

编写了有效的c++分配器后，必须将其作为共享指针传递给发布器、侦听器和执行程序。

auto alloc = std::make_shared<MyAllocator<void>>();
auto publisher = node->create_publisher<std_msgs::msg::UInt32>("allocator_example", 10, alloc);
auto msg_mem_strat =
  std::make_shared<rclcpp::message_memory_strategy::MessageMemoryStrategy<std_msgs::msg::UInt32,
  MyAllocator<>>>(alloc);
auto subscriber = node->create_subscription<std_msgs::msg::UInt32>(
  "allocator_example", 10, callback, nullptr, false, msg_mem_strat, alloc);

std::shared_ptr<rclcpp::memory_strategy::MemoryStrategy> memory_strategy =
  std::make_shared<AllocatorMemoryStrategy<MyAllocator<>>>(alloc);
rclcpp::executors::SingleThreadedExecutor executor(memory_strategy);

你还需要使用分配器分配沿着执行代码路径传递的任何消息。

auto alloc = std::make_shared<MyAllocator<void>>();

一旦你实例化了节点并将执行器添加到该节点，就该开始旋转了:

uint32_t i = 0;
while (rclcpp::ok()) {
    
  msg->data = i;
  i++;
  publisher->publish(msg);
  rclcpp::utilities::sleep_for(std::chrono::milliseconds(1));
  executor.spin_some();
}

4.将分配器传递给进程内部管道

即使我们在同一个进程中实例化了发布器和侦听器，我们还没有使用进程内通道。

IntraProcessManager是一个通常对用户隐藏的类，但是为了将自定义分配器传递给它，我们需要通过从rclcpp环境获取它，来公开它。IntraProcessManager使用了几个标准库结构，因此如果没有自定义分配器，它将调用默认新的。

auto context = rclcpp::contexts::default_context::get_global_default_context();
auto ipm_state =
  std::make_shared<rclcpp::intra_process_manager::IntraProcessManagerState<MyAllocator<>>>();
// Constructs the intra-process manager with a custom allocator.
context->get_sub_context<rclcpp::intra_process_manager::IntraProcessManager>(ipm_state);
auto node = rclcpp::Node::make_shared("allocator_example", true);

确保在以这种方式构造节点后，实例化发布器和侦听器。

5.测试和验证代码

你如何知道，你的自定义分配器实际上正在被调用?

显而易见的做法是计算对自定义分配器的allocate和deallocate函数的调用（次数），并将其与对new和delete的调用（次数）进行比较。

向自定义分配器添加计数（功能是）很简单:

T * allocate(size_t size, const void * = 0) {
    
  // ...
  num_allocs++;
  // ...
}

void deallocate(T * ptr, size_t size) {
    
  // ...
  num_deallocs++;
  // ...
}

你也可以覆盖全局new和delete操作符:

void operator delete(void * ptr) noexcept {
    
  if (ptr != nullptr) {
    
    if (is_running) {
    
      global_runtime_deallocs++;
    }
    std::free(ptr);
    ptr = nullptr;
  }
}

void operator delete(void * ptr, size_t) noexcept {
    
  if (ptr != nullptr) {
    
    if (is_running) {
    
      global_runtime_deallocs++;
    }
    std::free(ptr);
    ptr = nullptr;
  }
}

其中，我们递增的变量只是全局静态整数，而is_running是一个全局静态布尔值，在调用spin之前被切换。

示例（https://github.com/ros2/demos/blob/foxy/demo_nodes_cpp/src/topics/allocator_tutorial.cpp）可执行文件打印变量的值。要运行示例可执行文件，请使用:

allocator_example

或者，使用进程内管道运行示例:

allocator_example intra-process

你应该得到这样的数字:

Global new was called 15590 times during spin
Global delete was called 15590 times during spin
Allocator new was called 27284 times during spin
Allocator delete was called 27281 times during spin

我们已经捕获了发生在执行路径上的大约2/3的分配/释放，但是剩下的1/3来自哪里呢?

事实上，本例使用底层DDS实现这些分配/释放（操作）。

证明这是超出了本教程的范围,但是你可以查看配置的测试路径,运行的ROS2持续集成测试,通过代码和数据回溯跟踪，（看看）调用特定的函数是否是由DDS或者rmw来实现:

https://github.com/ros2/realtime_support/blob/foxy/tlsf_cpp/test/test_tlsf.cpp#

注意，这个测试没有使用我们刚刚创建的自定义分配器，而是使用TLSF分配器(见下面)。

6.TLSF分配器

ROS2支持TLSF (Two Level Segregate Fit)分配器，其设计是为了满足实时需求:

https://github.com/ros2/realtime_support/tree/foxy/tlsf_cpp

有关TLSF的更多信息，请参见

http://www.gii.upv.es/tlsf/

注意，TLSF分配器是在双gpl /LGPL许可证下许可的。

使用TLSF分配器的完整示例如下:

https://github.com/ros2/realtime_support/blob/foxy/tlsf_cpp/example/allocator_example.cpp