3. 共享内存传输配置

共享内存传输配置 学习卡片

任务目标

在DDS（以Fast DDS为例）中启用共享内存（Shared Memory）传输，替代默认的UDP/TCP传输，并完成与TCP/UDP传输的端到端延迟对比，验证共享内存在同机通信场景下的性能优势。

环境与依赖

  • Ubuntu 20.04+ / Windows 10+，Fast DDS 2.6+（本文以2.6.0为例）
  • 编译工具：CMake 3.16+，g++ 7+ 或 MSVC 2019+
  • 安装Fast DDS：sudo apt install ros-foxy-rmw-fastrtps-cpp（ROS 2环境）或从源码编译

核心实现

共享内存传输通过Fast DDS的SharedMemTransportDescriptor配置。关键点：共享内存传输仅在同一台机器的进程间有效，跨机器通信时需回退到UDP。以下代码实现一个可切换传输类型的发布-订阅对，通过环境变量选择传输方式。

// transport_compare.cpp
#include <fastdds/dds/domain/DomainParticipantFactory.hpp>
#include <fastdds/dds/domain/DomainParticipant.hpp>
#include <fastdds/dds/topic/TypeSupport.hpp>
#include <fastdds/dds/pub/Publisher.hpp>
#include <fastdds/dds/sub/Subscriber.hpp>
#include <fastdds/dds/sub/DataReader.hpp>
#include <fastdds/rtps/transport/shared_mem/SharedMemTransportDescriptor.hpp>
#include <fastdds/rtps/transport/UDPv4TransportDescriptor.hpp>
#include <fastdds/rtps/transport/TCPv4TransportDescriptor.hpp>
#include <fastdds/dds/sub/qos/DataReaderQos.hpp>
#include <fastdds/dds/pub/qos/DataWriterQos.hpp>
#include <atomic>
#include <chrono>
#include <iostream>
#include <thread>

using namespace eprosima::fastdds::dds;
using namespace eprosima::fastdds::rtps;

// 消息类型（简化，实际需用fastddsgen生成）
struct ChatMessage { std::string text; };
class ChatMessagePubSubType : public TopicDataType {
public:
    ChatMessagePubSubType() { setName("ChatMessage"); }
    bool serialize(void* data, SerializedPayload_t* payload) override {
        auto* msg = static_cast<ChatMessage*>(data);
        payload->length = msg->text.size() + 1;
        payload->data = (octet*)malloc(payload->length);
        memcpy(payload->data, msg->text.c_str(), payload->length);
        return true;
    }
    bool deserialize(SerializedPayload_t* payload, void* data) override {
        auto* msg = static_cast<ChatMessage*>(data);
        msg->text.assign((char*)payload->data, payload->length - 1);
        return true;
    }
    uint32_t calculate_serialized_size(void*, DataRepresentationId_t) override { return 0; }
    void* createData() override { return new ChatMessage(); }
    void deleteData(void* data) override { delete static_cast<ChatMessage*>(data); }
    bool getKey(void*, InstanceHandle_t*, bool) override { return false; }
};

std::atomic<bool> g_received{false};
std::atomic<long> g_latency_us{0};

class SubListener : public DataReaderListener {
    void on_data_available(DataReader* reader) override {
        ChatMessage msg;
        SampleInfo info;
        if (reader->take_next_sample(&msg, &info) == ReturnCode_t::RETCODE_OK) {
            if (info.valid_data) {
                auto now = std::chrono::steady_clock::now();
                auto send_time = std::chrono::steady_clock::time_point(
                    std::chrono::microseconds(std::stol(msg.text)));
                g_latency_us = std::chrono::duration_cast<std::chrono::microseconds>(now - send_time).count();
                g_received = true;
            }
        }
    }
};

void configure_transport(DomainParticipantQos& qos, const std::string& type) {
    if (type == "shm") {
        auto shm_transport = std::make_shared<SharedMemTransportDescriptor>();
        shm_transport->segment_size(1024 * 1024);  // 1MB共享内存段
        qos.transport().user_transports.push_back(shm_transport);
        qos.transport().use_builtin_transports = false;
    } else if (type == "udp") {
        auto udp_transport = std::make_shared<UDPv4TransportDescriptor>();
        qos.transport().user_transports.push_back(udp_transport);
        qos.transport().use_builtin_transports = false;
    } else if (type == "tcp") {
        auto tcp_transport = std::make_shared<TCPv4TransportDescriptor>();
        tcp_transport->add_listener_port(5100);
        qos.transport().user_transports.push_back(tcp_transport);
        qos.transport().use_builtin_transports = false;
    }
}

int main(int argc, char** argv) {
    if (argc < 3) {
        std::cerr << "Usage: " << argv[0] << " <pub|sub> <shm|udp|tcp>" << std::endl;
        return 1;
    }
    std::string role = argv[1];
    std::string transport = argv[2];

    DomainParticipantQos pqos;
    pqos.name("participant_" + role);
    configure_transport(pqos, transport);

    DomainParticipant* participant = DomainParticipantFactory::get_instance()->create_participant(0, pqos);
    if (!participant) { std::cerr << "Participant creation failed" << std::endl; return 1; }

    ChatMessagePubSubType type;
    TypeSupport ts(&type);
    participant->register_type(ts);

    Topic* topic = participant->create_topic("chat_topic", "ChatMessage", TOPIC_QOS_DEFAULT);

    if (role == "pub") {
        Publisher* pub = participant->create_publisher(PUBLISHER_QOS_DEFAULT);
        DataWriter* writer = pub->create_datawriter(topic, DATAWRITER_QOS_DEFAULT);
        std::cout << "Publisher ready (" << transport << "). Sending 1000 messages..." << std::endl;
        for (int i = 0; i < 1000; ++i) {
            ChatMessage msg;
            auto send_time = std::chrono::steady_clock::now();
            msg.text = std::to_string(
                std::chrono::duration_cast<std::chrono::microseconds>(send_time.time_since_epoch()).count());
            writer->write(&msg);
            std::this_thread::sleep_for(std::chrono::milliseconds(1));
        }
        std::this_thread::sleep_for(std::chrono::seconds(1));
    } else {
        SubListener listener;
        Subscriber* sub = participant->create_subscriber(SUBSCRIBER_QOS_DEFAULT);
        DataReader* reader = sub->create_datareader(topic, DATAREADER_QOS_DEFAULT, &listener);
        std::cout << "Subscriber ready (" << transport << "). Waiting for messages..." << std::endl;
        int count = 0;
        while (count < 1000) {
            if (g_received) {
                count++;
                if (count % 100 == 0) {
                    std::cout << "Received " << count << " msgs, avg latency: "
                              << g_latency_us << " us" << std::endl;
                }
                g_received = false;
            }
            std::this_thread::sleep_for(std::chrono::milliseconds(1));
        }
    }
    participant->delete_contained_entities();
    DomainParticipantFactory::get_instance()->delete_participant(participant);
    return 0;
}

编译与运行

# CMakeLists.txt
cmake_minimum_required(VERSION 3.16)
project(transport_compare)
find_package(fastdds REQUIRED)
add_executable(transport_compare transport_compare.cpp)
target_link_libraries(transport_compare fastdds fastcdr)
mkdir build && cd build
cmake ..
make -j4

# 终端1：共享内存订阅者
./transport_compare sub shm
# 终端2：共享内存发布者
./transport_compare pub shm

# 对比测试：分别用 udp 和 tcp 替换 shm 重复上述步骤

验证方法与判定标准

  1. 功能验证：订阅端每收到100条消息输出一次延迟，最终收到全部1000条消息即通信正常。
  2. 性能对比：分别记录三种传输方式下最后输出的平均延迟值（微秒）。同机场景下预期结果：shm < udp < tcp。共享内存通常比UDP低30%-50%，比TCP低50%-70%（具体数值取决于机器负载，但排序关系稳定）。
  3. 判定通过：三种传输均完成1000条消息收发，且共享内存延迟明显低于UDP/TCP（至少低20%），即验证了共享内存传输的配置正确性和性能优势。

常见问题

  • 共享内存段大小不足：若消息较大，需增大segment_size，否则会回退到UDP传输（可通过日志观察）。
  • 跨机器通信失败：共享内存仅限本机，跨机器需使用UDP/TCP，这是正常现象。
  • TCP传输需配置监听端口：发布端和订阅端需使用相同的端口号（本例中为5100）。
同机DDS通信延迟对比（示意） 延迟(μs) 0 50 100 ~15μs Shared Memory ~35μs UDP ~55μs TCP 零拷贝 传输类型（同机、消息大小1KB）

上图展示了同机场景下三种传输方式的延迟量级关系（示意，非实测数据）。共享内存通过内核零拷贝机制避免了网络协议栈开销，因此延迟最低。实际数值会因消息大小、系统负载而不同，但性能排序关系稳定。

← 上一章 📖 返回目录 下一章 →