Skip to content

fix: migrate SharedTopicClient to raw topic API #1

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Jiu-xiao wants to merge 1 commit into
base: master
Choose a base branch
from
Open

fix: migrate SharedTopicClient to raw topic API #1

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 7 additions & 7 deletions README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,25 +23,25 @@ SharedTopicClient is a client module for multi-topic data sharing and transparen

`SharedTopicClient` 不创建发送线程。模块注册 Topic callback；每次 Topic 发布时，
callback 先从空槽位队列申请一个 packet 槽，完成打包后把 `{槽位, 长度}` 放入待发
pool，然后尝试交给 UART `WritePort`：
队列，然后尝试交给 UART `WritePort`：

- 所有 Topic 共用同一组固定 packet 槽位。
- 空槽位用 `LockFreeQueue<uint32_t>` 管理；TX 消费完成后把槽位还回队列。
- 待发 packet 用 `LockFreePool` 管理；message callback 打包完成后放入 pool。
- packet 总数、空槽队列深度、待发 pool 深度相同。
- 空槽位用 `MPMCQueue<uint32_t>` 管理；TX 消费完成后把槽位还回队列。
- 待发 packet 用 `MPMCQueue<ReadyPacket>` 管理；message callback 打包完成后按 FIFO 放入队列。
- packet 总数由 `slot_count` 决定；队列容量比 slot 数多 1，以适配 libxr MPMC 队列的容量约束。
- 空槽申请失败时丢弃当前新 packet；这是全局背压，不是同 Topic 覆盖。

`TxService()` 每次只尝试交付一个待发 packet，不单独维护发送锁；并发提交与写队列容量
由 libxr `WritePort` 负责。如果 `WritePort` 暂时忙或写队列已满，当前待发 packet
会放回待发 pool，等待写完成回调或下一次 Topic callback 再推进。这样不会为转发链路
额外引入发送线程，也不会重复实现 `WritePort` 已经具备的互斥语义。
会丢弃并释放当前 packet 槽，等待写完成回调或下一次 Topic callback 再推进。这样不会为
转发链路额外引入发送线程，也不会重复实现 `WritePort` 已经具备的互斥语义。

## Timestamp

`SharedTopicClient` 转发 Topic 时会保留 libxr message envelope timestamp：

1. 本地 Topic callback 收到 `(timestamp, payload)`。
2. `Topic::PackData(topic_crc, buffer, timestamp, payload)` 写入串口包。
2. `Topic(topic_handle).PackRaw(payload, buffer, timestamp)` 写入串口包。
3. 对端 `SharedTopic` 解析后用同一个 timestamp 发布到对端 domain。

因此同步类 topic 不需要在 payload 里重复携带时间戳；payload 只保留业务字段即可。
Expand Down
33 changes: 17 additions & 16 deletions SharedTopicClient.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,16 +19,15 @@ depends: []
#include <cstdint>

#include "app_framework.hpp"
#include "lockfree_pool.hpp"
#include "lockfree_queue.hpp"
#include "message.hpp"
#include "queue.hpp"
#include "uart.hpp"

class SharedTopicClient : public LibXR::Application {
private:
struct CallbackInfo {
SharedTopicClient* client;
uint32_t topic_crc32;
LibXR::Topic::TopicHandle topic;
};

struct PacketSlot {
Expand Down Expand Up @@ -72,15 +71,15 @@ class SharedTopicClient : public LibXR::Application {
ASSERT(false);
}
const size_t packet_size =
topic->data_.max_length + LibXR::Topic::PACK_BASE_SIZE;
LibXR::Topic(topic).PayloadSize() + LibXR::Topic::PACK_BASE_SIZE;
max_packet_size = LibXR::max(max_packet_size, packet_size);
}

ASSERT(max_packet_size <= uart_->write_port_->queue_data_->MaxSize());

packets_ = new PacketSlot[slot_count];
free_slots_ = new LibXR::LockFreeQueue<uint32_t>(slot_count + 1);
ready_packets_ = new LibXR::LockFreePool<ReadyPacket>(slot_count);
free_slots_ = new LibXR::MPMCQueue<uint32_t>(slot_count + 1);
ready_packets_ = new LibXR::MPMCQueue<ReadyPacket>(slot_count + 1);
for (uint32_t i = 0; i < slot_count; i++) {
packets_[i].buffer =
LibXR::RawData(new uint8_t[max_packet_size], max_packet_size);
Expand All @@ -99,14 +98,15 @@ class SharedTopicClient : public LibXR::Application {
auto topic_handle = LibXR::Topic::Find(config.name, &domain);
ASSERT(topic_handle != nullptr);
void (*func)(bool, CallbackInfo, LibXR::MicrosecondTimestamp,
LibXR::ConstRawData&) =
const LibXR::ConstRawData&) =
[](bool in_isr, CallbackInfo info,
LibXR::MicrosecondTimestamp timestamp, LibXR::ConstRawData& data) {
LibXR::MicrosecondTimestamp timestamp,
const LibXR::ConstRawData& data) {
info.client->OnTopic(in_isr, info, timestamp, data);
};

auto msg_cb = LibXR::Topic::Callback::Create(
func, CallbackInfo{this, topic_handle->data_.crc32});
auto msg_cb =
LibXR::Topic::Callback::Create(func, CallbackInfo{this, topic_handle});

LibXR::Topic topic(topic_handle);

Expand All @@ -121,7 +121,7 @@ class SharedTopicClient : public LibXR::Application {
private:
void OnTopic(bool in_isr, CallbackInfo info,
LibXR::MicrosecondTimestamp timestamp,
LibXR::ConstRawData& data) {
const LibXR::ConstRawData& data) {
const size_t packet_size = data.size_ + LibXR::Topic::PACK_BASE_SIZE;
uint32_t slot_index = 0;

Expand All @@ -131,9 +131,10 @@ class SharedTopicClient : public LibXR::Application {

auto& slot = packets_[slot_index];
ASSERT(packet_size <= slot.buffer.size_);
LibXR::Topic::PackData(info.topic_crc32, slot.buffer, timestamp, data);
auto topic = LibXR::Topic(info.topic);
ASSERT(topic.PackRaw(data, slot.buffer, timestamp) == LibXR::ErrorCode::OK);

if (ready_packets_->Put(ReadyPacket{slot_index, packet_size}) !=
if (ready_packets_->Push(ReadyPacket{slot_index, packet_size}) !=
LibXR::ErrorCode::OK) {
ReturnFreeSlot(slot_index);
return;
Expand All @@ -145,7 +146,7 @@ class SharedTopicClient : public LibXR::Application {

void TxService(bool in_isr) {
ReadyPacket packet;
if (ready_packets_->Get(packet) != LibXR::ErrorCode::OK) {
if (ready_packets_->Pop(packet) != LibXR::ErrorCode::OK) {
return;
}

Expand Down Expand Up @@ -174,8 +175,8 @@ class SharedTopicClient : public LibXR::Application {

LibXR::UART* uart_;
PacketSlot* packets_ = nullptr;
LibXR::LockFreeQueue<uint32_t>* free_slots_ = nullptr;
LibXR::LockFreePool<ReadyPacket>* ready_packets_ = nullptr;
LibXR::MPMCQueue<uint32_t>* free_slots_ = nullptr;
LibXR::MPMCQueue<ReadyPacket>* ready_packets_ = nullptr;
LibXR::Callback<LibXR::ErrorCode> tx_callback_;
LibXR::WriteOperation tx_op_;
};
Loading