refactor: parse shared topics in rx thread

Author: Jiu-xiao

README.md

@@ -21,13 +21,12 @@ SharedTopic is a UART-based multi-topic data sharing and server module.
 
 ## 运行方式
 
-`SharedTopic` 不创建接收线程。模块在 UART `read_port` 上挂 0 字节 read 作为可读事件；
-一旦 UART 收到数据，回调会读取当前已有字节并调用 `Topic::Server::ParseDataFromCallback()`。
-解析出的 Topic 会继续沿 libxr callback/ISR 语义发布，串口包里的 envelope timestamp 会被保留。
+`SharedTopic` 创建一个接收线程。线程用 UART `read_port` 的阻塞读等待可读事件，
+被唤醒后一次性搬走当前已经到达的字节，再整块调用 `Topic::Server::ParseData()`。
+解析出的 Topic 走普通 `Topic::Publish()` 语义，串口包里的 envelope timestamp 会被保留。
 
-读取 UART 时会先按 `buffer_size` 成块搬到本地缓存，再逐字节喂给 `Topic::Server`。
-这样可以避免 parser 内部固定队列在“半包残留 + 新批次较大”时按整批写入语义丢数据。
-`buffer_size` 仍必须不小于需要接收的最大单个 Topic 串口包。
+业务 Topic 不在 UART/USB 回调链里发布，避免接收回调里继续触发同步命令等业务逻辑。
+`buffer_size` 必须不小于需要接收的最大单个 Topic 串口包。
 
 ## 依赖 / Depends
 

SharedTopic.hpp

@@ -25,6 +25,7 @@ depends: []
 #include "libxr_type.hpp"
 #include "logger.hpp"
 #include "message.hpp"
+#include "semaphore.hpp"
 #include "thread.hpp"
 #include "uart.hpp"
 
@@ -43,15 +44,15 @@ class SharedTopic : public LibXR::Application {
   SharedTopic(LibXR::HardwareContainer& hw, LibXR::ApplicationManager& app,
               const char* uart_name, uint32_t buffer_size,
               std::initializer_list<TopicConfig> topic_configs)
-      : uart_(hw.template Find<LibXR::UART>(uart_name)),
+      : uart_(hw.template FindOrExit<LibXR::UART>({uart_name})),
         server_(buffer_size),
         rx_buffer_(new uint8_t[buffer_size], buffer_size),
         cmd_name_(new char[sizeof("shared_topic:") + strlen(uart_name)]),
         cmd_file_((strcpy(cmd_name_, "shared_topic:"),
                    strcpy(cmd_name_ + strlen("shared_topic:"), uart_name),
                    LibXR::RamFS::CreateFile(cmd_name_, CommandFunc, this))) {
-    ASSERT(uart_ != nullptr);
     ASSERT(uart_->read_port_ != nullptr);
+    ASSERT(uart_->read_port_->Readable());
 
     for (auto config : topic_configs) {
       auto domain = LibXR::Topic::Domain(config.domain);
@@ -65,13 +66,8 @@ class SharedTopic : public LibXR::Application {
 
     hw.template FindOrExit<LibXR::RamFS>({"ramfs"})->Add(cmd_file_);
 
-    rx_callback_ = LibXR::Callback<LibXR::ErrorCode>::CreateGuarded(
-        [](bool in_isr, SharedTopic* self, LibXR::ErrorCode status) {
-          self->OnRxReady(in_isr, status);
-        },
-        this);
-    rx_ready_op_ = LibXR::ReadOperation(rx_callback_);
-    StartRxWait(false);
+    rx_thread_.Create<SharedTopic*>(this, RxThread, "shared_topic", 2048,
+                                    LibXR::Thread::Priority::MEDIUM);
 
     app.Register(*this);
   }
@@ -81,7 +77,8 @@ class SharedTopic : public LibXR::Application {
   static int CommandFunc(SharedTopic* self, int argc, char** argv) {
     if (argc == 1) {
       LibXR::STDIO::Printf<"Usage:\r\n">();
-      LibXR::STDIO::Printf<"  monitor [time_ms] [interval_ms] - test received speed\r\n">();
+      LibXR::STDIO::Printf<
+          "  monitor [time_ms] [interval_ms] - test received speed\r\n">();
       return 0;
     } else if (argc == 4) {
       if (strcmp(argv[1], "monitor") == 0) {
@@ -90,8 +87,9 @@ class SharedTopic : public LibXR::Application {
         auto start = self->rx_count_;
         while (time > 0) {
           LibXR::Thread::Sleep(delay);
-          LibXR::STDIO::Printf<"%f Mbps\r\n">(static_cast<float>(self->rx_count_ - start) * 8.0 /
-                                 1024.0 / 1024.0 / delay * 1000.0);
+          LibXR::STDIO::Printf<"%f Mbps\r\n">(
+              static_cast<float>(self->rx_count_ - start) * 8.0 / 1024.0 /
+              1024.0 / delay * 1000.0);
           time -= delay;
           start = self->rx_count_;
         }
@@ -105,29 +103,29 @@ class SharedTopic : public LibXR::Application {
   }
 
  private:
-  void OnRxReady(bool in_isr, LibXR::ErrorCode status) {
-    if (status == LibXR::ErrorCode::OK) {
-      auto size = LibXR::min(uart_->read_port_->Size(), rx_buffer_.size_);
-      if (size > 0) {
-        auto ans =
-            uart_->Read(LibXR::RawData{rx_buffer_.addr_, size}, rx_data_op_, in_isr);
-        if (ans == LibXR::ErrorCode::OK) {
-          auto* data = static_cast<uint8_t*>(rx_buffer_.addr_);
-          for (size_t i = 0; i < size; i++) {
-            server_.ParseDataFromCallback(LibXR::ConstRawData{data + i, 1},
-                                          in_isr);
-          }
-          rx_count_ += size;
-        }
+  static void RxThread(SharedTopic* self) { self->RunRxLoop(); }
+
+  void RunRxLoop() {
+    LibXR::ReadOperation wait_op(rx_sem_);
+    LibXR::ReadOperation read_op;
+
+    while (true) {
+      auto read_status = uart_->Read({nullptr, 0}, wait_op);
+      if (read_status != LibXR::ErrorCode::OK) {
+        continue;
       }
-    }
 
-    StartRxWait(in_isr);
-  }
+      while (uart_->read_port_->Size() > 0) {
+        auto size = LibXR::min(uart_->read_port_->Size(), rx_buffer_.size_);
+        read_status = uart_->Read(LibXR::RawData{rx_buffer_.addr_, size}, read_op);
+        if (read_status != LibXR::ErrorCode::OK) {
+          break;
+        }
 
-  void StartRxWait(bool in_isr) {
-    auto ans = uart_->Read({nullptr, 0}, rx_ready_op_, in_isr);
-    ASSERT(ans == LibXR::ErrorCode::OK);
+        server_.ParseData(LibXR::ConstRawData{rx_buffer_.addr_, size});
+        rx_count_ += size;
+      }
+    }
   }
 
   LibXR::UART* uart_;
@@ -142,7 +140,6 @@ class SharedTopic : public LibXR::Application {
 
   LibXR::RamFS::File cmd_file_;
 
-  LibXR::Callback<LibXR::ErrorCode> rx_callback_;
-  LibXR::ReadOperation rx_ready_op_;
-  LibXR::ReadOperation rx_data_op_;
+  LibXR::Semaphore rx_sem_;
+  LibXR::Thread rx_thread_;
 };