DomainParticipantQos qos;

// configure an initial peer on host 192.168.10.13.
// The port number corresponds to the well-known port
for metatraffic unicast
// on participant ID `1` and domain `0`.
Locator_t initial_peer;
IPLocator::setIPv4(initial_peer, "192.168.10.13");
initial_peer.port = 7412;
qos.wire_protocol().builtin.initialPeersList.push_back(initial_peer);

// Get default participant QoS
DomainParticipantQos server_qos = PARTICIPANT_QOS_DEFAULT;

// Set participant as SERVER
server_qos.wire_protocol().builtin.discovery_config.
discoveryProtocol = DiscoveryProtocol_t::SERVER;

// Set SERVER's GUID prefix
std::istringstream("44.53.00.5f.45.50.52.4f.53.49.4d.41") >>
server_qos.wire_protocol().prefix;

// Set SERVER's listening locator for PDP
Locator_t locator;
IPLocator::setIPv4(locator, 127, 0, 0, 1);
locator.port = 11811;
server_qos.wire_protocol().builtin.metatrafficUnicastLocatorList.
push_back(locator);

/* Add a remote serve to which this server will connect */
// Set remote SERVER's GUID prefix
RemoteServerAttributes remote_server_att;
remote_server_att.ReadguidPrefix("44.53.01.5f.45.50.52.4f.53.49.4d.41");

// Set remote SERVER's listening locator for PDP
Locator_t remote_locator;
IPLocator::setIPv4(remote_locator, 127, 0, 0, 1);
remote_locator.port = 11812;
remote_server_att.metatrafficUnicastLocatorList.
push_back(remote_locator);

// Add remote SERVER to SERVER's list of SERVERs
server_qos.wire_protocol().builtin.discovery_config.m_Discovery
Servers.push_back(remote_server_att);

// Create SERVER
DomainParticipant* server =
          DomainParticipantFactory::get_instance()->create_participant(0, server_qos);
if (nullptr == server)
{
      // Error
       return;
}

# 查看操作系统socket缓冲区最大值
sudo sysctl -a | grep net.core.wmem_max # 发送
sudo sysctl -a | grep net.core.rmem_max # 接收

# 修改操作系统socket缓冲区最大值
sudo sysctl -w net.core.wmem_max=12582912 # 发送
sudo sysctl -w net.core.rmem_max=12582912 # 接收

# 修改dds默认socket buffer size
DomainParticipantQos participant_qos;
participant_qos.transport().send_socket_buffer_size =
1048576; // sending buffer size
participant_qos.transport().listen_socket_buffer_size =
4194304; // receiving buffer size

# 查看指定网卡的传输队列长度设置
ifconfig wlo1 | grep txqueuelen

# 修改
ifconfig ${interface} txqueuelen ${size}

// 限制300kb/s
static const char* flow_controller_name = "example_flow_controller";
auto flow_control_300k_per_sec = std::make_shared<eprosima::fastdds::rtps::FlowControllerDescriptor>();
flow_control_300k_per_sec->name = flow_controller_name;
flow_control_300k_per_sec->scheduler = eprosima::fastdds::rtps::FlowControllerSchedulerPolicy::FIFO;
flow_control_300k_per_sec->max_bytes_per_period = 300 * 1000;
flow_control_300k_per_sec->period_ms = 1000;

// 为域成员策略 配置流控制器
DomainParticipantQos participant_qos;
participant_qos.flow_controllers().push_back(flow_control_300k_per_sec);

// .... 创建域成员和发布者

// 为DataWriter策略关联流控制器
DataWriterQos qos;
qos.publish_mode().kind = ASYNCHRONOUS_PUBLISH_MODE;
qos.publish_mode().flow_controller_name = flow_controller_name;

DataWriterQos qos;
qos.reliable_writer_qos().times.heartbeatPeriod.seconds = 0;
qos.reliable_writer_qos().times.heartbeatPeriod.nanosec =
500000000; //500 ms

// 添加一个多播定位器（指定IP和端口列表）
eprosima::fastrtps::rtps::Locator_t new_multicast_locator;
eprosima::fastrtps::rtps::IPLocator::setIPv4(new_multicast_locator, "239.255.0.4");
new_multicast_locator.port = 7900;
qos.endpoint().multicast_locator_list.push_back(new_multicast_locator);

RTPSEndpointQos endpoint;
endpoint.history_memory_policy = eprosima::fastrtps::rtps::DYNAMIC_REUSABLE_MEMORY_MODE;

struct LoanableHelloWorld {
         unsigned long index;
         char message[256];
};

// 打开data_sharing的writer
DataWriterQos wqos = publisher_->get_default_datawriter_qos();
wqos.history().depth = 10;
wqos.durability().kind = TRANSIENT_LOCAL_DURABILITY_QOS;
wqos.data_sharing().automatic();

writer_ = publisher_->create_datawriter(topic_, wqos, &listener_);

void* sample = nullptr; // 借用缓冲区存储pub数据
if (ReturnCode_t::RETCODE_OK == writer_->loan_sample(sample)){
         std::cout << "Preparing sample at address " << sample <<          std::endl;
         LoanableHelloWorld* data =          static_cast<LoanableHelloWorld*>(sample);
         data->index() = msgsent + 1;
         memcpy(data->message().data(),
"LoanableHelloWorld ", 20);
         writer_->write(sample); // write后归还缓冲区的持有权。如果loan_sample()后不write，需要使用discard_loan()释放缓冲区
}

//CREATE THE READER
DataReaderQos rqos = subscriber_->get_default_datareader_qos();
rqos.history().depth = 10;
rqos.reliability().kind = RELIABLE_RELIABILITY_QOS;
rqos.durability().kind = TRANSIENT_LOCAL_DURABILITY_QOS;
rqos.data_sharing().automatic();

reader_ = subscriber_->create_datareader(topic_, rqos, &listener_);

void LoanableHelloWorldSubscriber::SubListener:
:on_data_available(DataReader* reader)
{
         FASTDDS_CONST_SEQUENCE(DataSeq, LoanableHelloWorld);

         DataSeq data;
         SampleInfoSeq infos;
         while (ReturnCode_t::RETCODE_OK == reader->take(data, infos))
         {
            for (LoanableCollection::size_type i = 0; i < infos.length(); ++i)
               {
                  if (infos[i].valid_data)
             {
                  // 通过接收队列获得样本数据的引用
                  const LoanableHelloWorld& sample = data[i];

                  std::cout << "Sample received (count=" <<
samples << ") at  address " << &sample
                   <<(reader->is_sample_valid(&sample, &
infos[i]) ? " is valid"  was replaced" ) << std::endl
                           << " index=" << sample.index() << std::endl
                          << " message=" <<
sample.message().data() << std::endl;
                          }
              }
             // 归还缓冲区持有权
              reader->return_loan(data, infos);
             }
}

cd ~/project/Fast-DDS vim src/fastrtps/CMakeLists.txt                           option(FASTDDS_STATISTICS
"Enable Fast DDS Statistics Module" ON) colcon build

# 统计DataWriters时延
export FASTDDS_STATISTICS="HISTORY_LATENCY_TOPIC;
NETWORK_LATENCY_TOPIC"

cd ~/project/FastDDS
vim fastrtps.repos
             fastdds_statistics_backend:
                          type: git
                           url: https://github.com/eProsima/Fast-DDS-statistics-backend.git
                          version: main

# download code
vcs import src < fastrtps.repos

# build
colcon build --cmake-target install
or
cd src/fastdds_statistics_backend/build
cmake ..
sudo cmake --build . --target install

# example
cd src/fastdds_statistics_backend/examples/cpp/
HelloWorldExample/build
cmake ..
cmake --build .

# 统计pub/sub的通信的延迟和吞吐量
./HelloWorldExample publisher
./HelloWorldExample subcriber
./HelloWorldExample monitor

Fast DDS Latency of HelloWorld topic: [ 206.079 μs]
Publication throughput of Participant Participant_pub: [ 229.257 B/s]

#include <fastdds_statistics_backend/StatisticsBackend.hpp>

using namespace eprosima::statistics_backend;

...

# 初始化监控
monitor_id_ = StatisticsBackend::init_monitor(domain);
if (!monitor_id_.is_valid()){
             std::cout << "Error creating monitor" << std::endl;
              return 1;
}
StatisticsBackend::set_physical_listener(&physical_listener_);

...

# 获取要监控的topicid
             std::vector<StatisticsData> latency_data{};

             std::vector<EntityId> topics = StatisticsBackend::get_entities(EntityKind::TOPIC);
              EntityId helloworld_topic_id = -1;
             Info topic_info;
             for (auto topic_id : topics)
             {
                          topic_info = StatisticsBackend::get_info(topic_id);
                          if (topic_info["name"] == "HelloWorldTopic" && topic_info["data_type"] == "HelloWorld")
                          {
                          helloworld_topic_id = topic_id;
             }
}

             if (helloworld_topic_id < 0)
             {
                          return latency_data;
             }

# 通过topic id获得对应的DataWriter和DataReader
std::vector<EntityId> topic_datawriters = StatisticsBackend::get_entities(EntityKind::DATAWRITER, helloworld_topic_id);
std::vector<EntityId> topic_datareaders = StatisticsBackend::get_entities(EntityKind::DATAREADER, helloworld_topic_id);

# 获得指定DataWriters和DataReaders间的传输时延统计数据
std::vector<StatisticsData> latency_data{};
std::chrono::system_clock::time_point now = std::chrono::system_clock::now();
latency_data = StatisticsBackend::get_data(
             DataKind::FASTDDS_LATENCY, // DataKind
             topic_datawriters, // Source entities
             topic_datareaders, // Target entities
             1, // Number of bins
             now - std::chrono::seconds(5), // t_from
              now, // t_to
              StatisticKind::MEAN); // Statistic

         for (auto latency : latency_data){
             std::cout << "Fast DDS Latency of HelloWorld topic: ["
                           << timestamp_to_string(latency.first) << ", " latency.second / 1000 << " μs]" << std::endl;
             }