1. QoS策略配置与性能基准测试

QoS策略配置与性能基准测试 学习卡片

任务目标

在DDS（Data Distribution Service）通信中，QoS（Quality of Service）策略直接决定数据分发的行为特征。本案例通过配置两种关键QoS策略——可靠性（Reliability）与历史记录深度（History Depth），测量它们对端到端延迟（Latency）和吞吐量（Throughput）的影响，建立QoS参数与性能指标之间的量化关系。

环境与依赖

  • 操作系统：Ubuntu 20.04+ / Windows 10+（需支持C++11）
  • 中间件：Eclipse Cyclone DDS（开源，支持C++ API）
  • 编译工具：CMake ≥ 3.10，支持C++11的编译器（GCC/Clang/MSVC）
  • 安装依赖：sudo apt install libcyclonedds-dev（Ubuntu）或从官方仓库源码编译

实现方案

设计一个发布-订阅（Pub-Sub）基准测试程序：发布端以固定频率发送带时间戳的小型数据包，订阅端接收后计算单包延迟（接收时刻 - 发送时刻），并统计累计吞吐量（每秒接收字节数）。通过命令行参数切换QoS配置组合，对比不同策略下的性能差异。

核心测量逻辑：

  • 延迟：每个数据包携带发送端单调时钟时间戳，接收端用同一时钟计算差值，取滑动窗口平均值。
  • 吞吐量：接收端每秒统计收到的字节总数，输出瞬时值。

关键完整代码

// qos_bench.cpp
#include <cstdlib>
#include <iostream>
#include <string>
#include <thread>
#include <chrono>
#include <atomic>
#include <cstring>
#include <vector>
#include "dds/dds.hpp"

using namespace dds::core;
using namespace dds::pub;
using namespace dds::sub;
using namespace dds::topic;
using namespace std::chrono;

// 数据类型定义
struct BenchData {
    uint64_t seq;
    uint64_t send_time_ns;
};
// 注册类型（Cyclone DDS要求）
namespace dds { namespace topic {
    template<> struct topic_type_support<BenchData> {
        static std::vector<uint8_t> to_bytes(const BenchData& d) {
            std::vector<uint8_t> buf(sizeof(BenchData));
            std::memcpy(buf.data(), &d, sizeof(BenchData));
            return buf;
        }
        static BenchData from_bytes(const std::vector<uint8_t>& buf) {
            BenchData d;
            std::memcpy(&d, buf.data(), sizeof(BenchData));
            return d;
        }
        static std::string type_name() { return "BenchData"; }
    };
}}

// 全局统计
std::atomic<uint64_t> g_recv_count{0};
std::atomic<uint64_t> g_recv_bytes{0};
std::atomic<uint64_t> g_latency_sum_ns{0};

int main(int argc, char** argv) {
    // 解析参数: mode(0=pub,1=sub) reliability(0=best_effort,1=reliable) depth(1/10/100)
    if (argc < 4) {
        std::cerr << "Usage: " << argv[0] << " <mode:0/1> <reliability:0/1> <depth:int>\n";
        return 1;
    }
    int mode = std::atoi(argv[1]);
    bool reliable = std::atoi(argv[2]) != 0;
    int depth = std::atoi(argv[3]);

    // 创建DomainParticipant（Domain ID=0）
    dds::domain::DomainParticipant dp(0);
    // 注册Topic
    dds::topic::Topic<BenchData> topic(dp, "bench_topic");
    // 构造QoS
    dds::pub::qos::DataWriterQos wqos;
    dds::sub::qos::DataReaderQos rqos;
    if (reliable) {
        wqos.reliability(dds::core::policy::Reliability::Reliable());
        rqos.reliability(dds::core::policy::Reliability::Reliable());
    } else {
        wqos.reliability(dds::core::policy::Reliability::BestEffort());
        rqos.reliability(dds::core::policy::Reliability::BestEffort());
    }
    wqos.history(dds::core::policy::History::KeepLast(depth));
    rqos.history(dds::core::policy::History::KeepLast(depth));

    if (mode == 0) { // 发布端
        dds::pub::Publisher pub(dp);
        dds::pub::DataWriter<BenchData> writer(pub, topic, wqos);
        BenchData sample;
        sample.seq = 0;
        auto start = high_resolution_clock::now();
        while (true) {
            sample.send_time_ns = duration_cast<nanoseconds>(
                high_resolution_clock::now().time_since_epoch()).count();
            writer.write(sample);
            sample.seq++;
            // 控制发送频率约10kHz（100us间隔）
            std::this_thread::sleep_for(microseconds(100));
            // 每2秒打印一次发送速率
            auto now = high_resolution_clock::now();
            if (duration_cast<seconds>(now - start).count() >= 2) {
                std::cout << "[PUB] sent=" << sample.seq << " rate=" 
                          << sample.seq / 2.0 << " msg/s\n";
                start = now;
                sample.seq = 0;
            }
        }
    } else { // 订阅端
        dds::sub::Subscriber sub(dp);
        dds::sub::DataReader<BenchData> reader(sub, topic, rqos);
        auto start = high_resolution_clock::now();
        uint64_t window_bytes = 0;
        while (true) {
            auto samples = reader.take();
            for (auto& s : samples) {
                if (s.info().valid()) {
                    auto now_ns = duration_cast<nanoseconds>(
                        high_resolution_clock::now().time_since_epoch()).count();
                    uint64_t latency = now_ns - s.data().send_time_ns;
                    g_latency_sum_ns += latency;
                    g_recv_count++;
                    g_recv_bytes += sizeof(BenchData);
                    window_bytes += sizeof(BenchData);
                }
            }
            auto now = high_resolution_clock::now();
            if (duration_cast<seconds>(now - start).count() >= 2) {
                double avg_lat_us = (g_recv_count > 0) ? 
                    (g_latency_sum_ns / (double)g_recv_count / 1000.0) : 0.0;
                double throughput_mbps = (window_bytes * 8) / 2.0 / 1e6;
                std::cout << "[SUB] count=" << g_recv_count 
                          << " avg_latency_us=" << avg_lat_us
                          << " throughput_mbps=" << throughput_mbps << "\n";
                // 重置窗口
                g_recv_count = 0;
                g_latency_sum_ns = 0;
                window_bytes = 0;
                start = now;
            }
            std::this_thread::sleep_for(milliseconds(1));
        }
    }
    return 0;
}

编译与运行

# CMakeLists.txt
cmake_minimum_required(VERSION 3.10)
project(qos_bench)
find_package(CycloneDDS REQUIRED)
add_executable(qos_bench qos_bench.cpp)
target_link_libraries(qos_bench CycloneDDS::CycloneDDS)
mkdir build && cd build
cmake ..
make -j4
# 终端1：发布端（BestEffort, depth=1）
./qos_bench 0 0 1
# 终端2：订阅端（BestEffort, depth=1）
./qos_bench 1 0 1

验证方法与判定标准

按以下矩阵运行4组实验，每组运行至少30秒，记录订阅端输出的平均延迟与吞吐量：

组号ReliabilityHistory Depth预期观察
ABestEffort1最低延迟，可能丢包
BReliable1延迟略高，无丢包
CReliable10延迟与B接近，容忍短暂网络抖动
DReliable100内存占用增加，延迟可能因缓存而略升

通过条件：

  1. 程序能稳定运行，订阅端每2秒输出统计信息，无崩溃或死锁。
  2. 对比A组与B组：A组平均延迟应显著低于B组（通常差距>20%），证明可靠性QoS引入额外确认开销。
  3. 对比B/C/D组：延迟差异应在测量噪声范围内（<15%），证明历史深度对稳态延迟影响有限，但深度越大，订阅端内存占用越高（可通过ps -o rss观察）。
  4. 所有组吞吐量应接近发布速率（约10k msg/s × 24B ≈ 1.9 Mbps），若Reliable组吞吐量明显下降，说明网络或中间件存在瓶颈。

注意：以上预期基于典型局域网环境，实际数值受硬件、网络和系统负载影响。若A组与B组延迟差异不明显，请检查是否在同一台机器上运行（回环接口延迟极低），建议在两台物理机或使用tc模拟网络延迟（sudo tc qdisc add dev lo root netem delay 1ms）以放大差异。

常见问题

  • 编译错误：找不到CycloneDDS：确认已安装libcyclonedds-dev，且CMake能找到CycloneDDSConfig.cmake。
  • 订阅端收不到数据：检查发布端和订阅端的Domain ID是否一致（本例均为0），且Topic名称相同（"bench_topic"）。
  • 延迟测量不准确：确保发布端和订阅端使用同一台机器的单调时钟（steady_clock），跨机器时需使用NTP同步或硬件时间戳。
← 上一章 📖 返回目录 下一章 →