Zero-Copy Communication

1
#include <fastdds/dds/domain/DomainParticipant.hpp>
2
#include <fastdds/dds/publisher/DataWriter.hpp>
3
#include <fastdds/dds/publisher/qos/DataWriterQos.hpp>
4

5
// 创建 Participant 和 Publisher（省略 Topic 创建代码）
6
// ...
7

8
// 关键：配置 Data-sharing 为 AUTO 或 ON
9
DataWriterQos wqos = publisher->get_default_datawriter_qos();
10
wqos.history().depth = 10;
11
wqos.durability().kind = TRANSIENT_LOCAL_DURABILITY_QOS;
12
wqos.data_sharing().automatic();  // 开启 Data-sharing
13

14
DataWriter* writer = publisher->create_datawriter(topic, wqos);
15

16
// Zero-Copy 发送流程
17
void* sample = nullptr;
18
if (RETCODE_OK == writer->loan_sample(sample))
19
{
20
    // 此时 sample 指向共享内存中的缓冲区
21
    // 直接在其中写入数据，无需任何拷贝
22
    LoanablePointCloud* data = static_cast<LoanablePointCloud*>(sample);
23
    data->timestamp() = get_current_timestamp();
24
    data->point_count() = 65536;
25
    fill_point_data(data->x(), data->y(), data->z());
26

27
    // 发布数据。调用后，缓冲区所有权归还中间件
28
    writer->write(sample);
29
}

1
// 关键：配置 Data-sharing 为 AUTO 或 ON
2
DataReaderQos rqos = subscriber->get_default_datareader_qos();
3
rqos.history().depth = 10;
4
rqos.reliability().kind = RELIABLE_RELIABILITY_QOS;
5
rqos.durability().kind = TRANSIENT_LOCAL_DURABILITY_QOS;
6
rqos.data_sharing().automatic();  // 开启 Data-sharing
7

8
DataReader* reader = subscriber->create_datareader(topic, rqos, &listener);
9

10
// Zero-Copy 接收流程（在 on_data_available 回调中）
11
void on_data_available(DataReader* reader)
12
{
13
    LoanableCollection<LoanablePointCloud> data;
14
    SampleInfoSeq infos;
15

16
    // 以 loan 方式读取数据
17
    // data 中的样本直接指向共享内存，无拷贝
18
    while (RETCODE_OK == reader->take(data, infos))
19
    {
20
        for (int i = 0; i < data.length(); ++i)
21
        {
22
            if (infos[i].valid_data)
23
            {
24
                // 直接访问共享内存中的数据
25
                const LoanablePointCloud& point = data[i];
26

27
                // 检查样本是否被覆盖（Data-sharing 的 history 耦合机制）
28
                // 如果 DataWriter 复用了同一个缓冲区，样本可能已失效
29
                bool valid = reader->is_sample_valid(&point, &infos[i]);
30
                if (valid)
31
                {
32
                    process_point_cloud(point.timestamp(), point.point_count(),
33
                                        point.x(), point.y(), point.z());
34
                }
35
            }
36
        }
37

38
        // 必须归还 loan！否则共享内存缓冲区不会释放
39
        reader->return_loan(data, infos);
40
    }
41
}