[BugFix][android][ios][common] use WorkThreadExecutor instead of raw thread

popoaichuiniu · RovicnowYoyi · commit cd4fedd22c41 · 2025-04-16T19:42:35.000+08:00
Using WorkThreadExecutor can conveniently stop all threads of usb_client when it ends, avoiding unexpected resource access. issue: lynx-family#46
diff --git a/debug_router/native/base/socket_guard.h b/debug_router/native/base/socket_guard.h
@@ -20,15 +20,20 @@ constexpr SocketType kInvalidSocket = INVALID_SOCKET;
 typedef int SocketType;
 constexpr SocketType kInvalidSocket = -1;
 #endif
+#include <mutex>
 
 namespace debugrouter {
 namespace base {
 
 class SocketGuard {
  public:
-  SocketType Get() const { return sock_; }
+  SocketType Get() {
+    std::lock_guard<std::mutex> lock(mutex_);
+    return sock_;
+  }
 
   void Reset() {
+    std::lock_guard<std::mutex> lock(mutex_);
     if (sock_ != kInvalidSocket) {
       CLOSESOCKET(sock_);
     }
@@ -37,16 +42,14 @@ class SocketGuard {
 
   explicit SocketGuard(SocketType sock) : sock_(sock) {}
 
-  ~SocketGuard() {
-    if (sock_ != kInvalidSocket) {
-      CLOSESOCKET(sock_);
-    }
-  }
+  ~SocketGuard() { Reset(); }
+
   SocketGuard(const SocketGuard&) = delete;
   SocketGuard& operator=(const SocketGuard&) = delete;
 
  private:
   SocketType sock_;
+  std::mutex mutex_;
 };
 
 }  // namespace base
diff --git a/debug_router/native/socket/usb_client.cc b/debug_router/native/socket/usb_client.cc
@@ -45,7 +45,7 @@ void UsbClient::CloseClientSocket(SocketType socket_fd_) {
   socket_fd_ = kInvalidSocket;
 }
 
-UsbClient::UsbClient(SocketType socket_fd) : socket_fd_(socket_fd) {
+UsbClient::UsbClient(SocketType socket_fd) : socket_guard_(socket_fd) {
   LOGI("UsbClient: Constructor.");
 }
 
@@ -55,7 +55,12 @@ void UsbClient::SetConnectStatus(USBConnectStatus status) {
   });
 }
 
-void UsbClient::Init() { work_thread_.init(); }
+void UsbClient::Init() {
+  work_thread_.init();
+  read_thread_.init();
+  write_thread_.init();
+  dispatch_thread_.init();
+}
 
 void UsbClient::StartUp(const std::shared_ptr<UsbClientListener> &listener) {
   LOGI("UsbClient: StartUp.");
@@ -69,9 +74,8 @@ void UsbClient::StartInternal(
   LOGI("UsbClient: StartInternal.");
   connect_status_ = USBConnectStatus::CONNECTING;
   listener_ = listener;
-  latch_ = std::make_unique<CountDownLatch>(kThreadCount);
-  StartReader(socket_fd_);
-  StartWriter(socket_fd_);
+  StartReader();
+  StartWriter();
 }
 
 bool UsbClient::ReadAndCheckMessageHeader(char *header, SocketType socket_fd_) {
@@ -132,14 +136,14 @@ bool UsbClient::Read(SocketType socket_fd_, char *buffer, uint32_t read_size) {
   return true;
 }
 
-void UsbClient::ReadMessage(SocketType socket_fd_) {
-  LOGI("UsbClient: ReadMessage:" << socket_fd_);
+void UsbClient::ReadMessage() {
+  LOGI("UsbClient: ReadMessage:" << socket_guard_.Get());
   bool isFirst = true;
   while (true) {
     char header[kFrameHeaderLen];
     memset(header, 0, kFrameHeaderLen);
     LOGI("UsbClient: start check message header.");
-    if (!ReadAndCheckMessageHeader(header, socket_fd_)) {
+    if (!ReadAndCheckMessageHeader(header, socket_guard_.Get())) {
       LOGW("UsbClient: don't match DebugRouter protocol:");
       // need DebugRouterReport to report invailed client.
       for (int i = 0; i < kFrameHeaderLen; i++) {
@@ -160,7 +164,7 @@ void UsbClient::ReadMessage(SocketType socket_fd_) {
       isFirst = false;
     }
     char payload_size[kPayloadSizeLen];
-    if (!Read(socket_fd_, payload_size, kPayloadSizeLen)) {
+    if (!Read(socket_guard_.Get(), payload_size, kPayloadSizeLen)) {
       LOGE("read payload data error: " << GetErrorMessage());
       if (listener_) {
         listener_->OnError(shared_from_this(), GetErrorMessage(),
@@ -181,7 +185,7 @@ void UsbClient::ReadMessage(SocketType socket_fd_) {
       continue;
     }
     char payload[payload_size_int];
-    if (!Read(socket_fd_, payload, payload_size_int)) {
+    if (!Read(socket_guard_.Get(), payload, payload_size_int)) {
       LOGE("read payload data error: " << GetErrorMessage());
       if (listener_) {
         listener_->OnError(shared_from_this(), GetErrorMessage(),
@@ -204,21 +208,13 @@ void UsbClient::ReadMessage(SocketType socket_fd_) {
   LOGI("UsbClient: ReadMessage thread exit.");
   incoming_message_queue_.put(std::move(kMessageQuit));
   outgoing_message_queue_.put(std::move(kMessageQuit));
-  if (latch_) {
-    latch_->CountDown();
-  }
 }
 
-void UsbClient::ReadThreadFunc(std::shared_ptr<UsbClient> client,
-                               SocketType socket_fd_) {
-  client->ReadMessage(socket_fd_);
-}
-
-void UsbClient::StartReader(SocketType socket_fd_) {
+void UsbClient::StartReader() {
   LOGI("UsbClient: start reader thread.");
-  StartMessageDispatcher(socket_fd_);
-  std::thread read_thread(ReadThreadFunc, shared_from_this(), socket_fd_);
-  read_thread.detach();
+  StartMessageDispatcher();
+  read_thread_.submit(
+      [client_ptr = shared_from_this()]() { client_ptr->ReadMessage(); });
 }
 
 void UsbClient::MessageDispatcher() {
@@ -240,20 +236,13 @@ void UsbClient::MessageDispatcher() {
     }
   }
   LOGI("UsbClient: message dispatcher finished.");
-  if (latch_) {
-    latch_->CountDown();
-  }
 }
 
-void UsbClient::MessageDispatcherFunc(std::shared_ptr<UsbClient> client) {
-  client->MessageDispatcher();
-}
-
-void UsbClient::StartMessageDispatcher(SocketType socket_fd_) {
+void UsbClient::StartMessageDispatcher() {
   LOGI("UsbClient: startMessageDispatcher.");
-  std::thread dispatch_message_thread(MessageDispatcherFunc,
-                                      shared_from_this());
-  dispatch_message_thread.detach();
+
+  dispatch_thread_.submit(
+      [client_ptr = shared_from_this()]() { client_ptr->MessageDispatcher(); });
 }
 
 void UsbClient::WrapHeader(const std::string &message, std::string &result) {
@@ -288,8 +277,8 @@ void UsbClient::WrapHeader(const std::string &message, std::string &result) {
   memcpy(buffer + 20, message.c_str(), message.size());
 }
 
-void UsbClient::WriteMessage(SocketType socket_fd_) {
-  LOGI("UsbClient: WriteMessage:" << socket_fd_);
+void UsbClient::WriteMessage() {
+  LOGI("UsbClient: WriteMessage:" << socket_guard_.Get());
   while (true) {
     std::string message;
     message = outgoing_message_queue_.take();
@@ -303,8 +292,8 @@ void UsbClient::WriteMessage(SocketType socket_fd_) {
       LOGI(message);
       std::string result_message;
       WrapHeader(message, result_message);
-      if (send(socket_fd_, result_message.c_str(), result_message.size(), 0) ==
-          -1) {
+      if (send(socket_guard_.Get(), result_message.c_str(),
+               result_message.size(), 0) == -1) {
         LOGE("send error: " << GetErrorMessage() << " message:" << message);
         if (listener_) {
           listener_->OnError(shared_from_this(), GetErrorMessage(),
@@ -322,38 +311,19 @@ void UsbClient::WriteMessage(SocketType socket_fd_) {
                        "writer thread finished");
   }
   LOGI("UsbClient: WriteMessage thread exit.");
-  if (latch_) {
-    latch_->CountDown();
-  }
 }
 
-void UsbClient::WriteThreadFunc(std::shared_ptr<UsbClient> client,
-                                SocketType socket_fd_) {
-  client->WriteMessage(socket_fd_);
-}
-
-void UsbClient::StartWriter(SocketType socket_fd_) {
+void UsbClient::StartWriter() {
   LOGI("UsbClient: start writer thread.");
-  std::thread write_thread(WriteThreadFunc, shared_from_this(), socket_fd_);
-  write_thread.detach();
+  write_thread_.submit(
+      [client_ptr = shared_from_this()]() { client_ptr->WriteMessage(); });
 }
 
 void UsbClient::DisconnectInternal() {
   LOGI("UsbClient: DisconnectInternal.");
-  if (latch_) {
-    incoming_message_queue_.put(std::move(kMessageQuit));
-    outgoing_message_queue_.put(std::move(kMessageQuit));
-
-    LOGI("UsbClient: DisconnectInternal waiting for threads exit.");
-    latch_->Await();
-    connect_status_ = USBConnectStatus::DISCONNECTED;
-
-    incoming_message_queue_.clear();
-    outgoing_message_queue_.clear();
-    latch_ = nullptr;
-    LOGI("UsbClient: DisconnectInternal successfully.");
-  }
-  CloseClientSocket(socket_fd_);
+  incoming_message_queue_.put(std::move(kMessageQuit));
+  outgoing_message_queue_.put(std::move(kMessageQuit));
+  socket_guard_.Reset();
 }
 
 bool UsbClient::Send(const std::string &message) {
@@ -369,7 +339,16 @@ bool UsbClient::Send(const std::string &message) {
   return true;
 }
 
-void UsbClient::Stop() { work_thread_.shutdown(); }
+void UsbClient::Stop() {
+  DisconnectInternal();
+  dispatch_thread_.shutdown();
+  write_thread_.shutdown();
+  read_thread_.shutdown();
+  work_thread_.shutdown();
+  incoming_message_queue_.clear();
+  outgoing_message_queue_.clear();
+  connect_status_ = USBConnectStatus::DISCONNECTED;
+}
 
 void UsbClient::SendInternal(const std::string &message) {
   LOGI("UsbClient: SendInternal.");
@@ -383,8 +362,7 @@ void UsbClient::SendInternal(const std::string &message) {
 
 UsbClient::~UsbClient() {
   LOGI("UsbClient: ~UsbClient.");
-  // TODO(popoaichuiniu) optimize thread policy of UsbClient's fields
-  DisconnectInternal();
+  Stop();
 }
 
 }  // namespace socket_server
diff --git a/debug_router/native/socket/usb_client.h b/debug_router/native/socket/usb_client.h
@@ -5,6 +5,7 @@
 #ifndef DEBUGROUTER_NATIVE_SOCKET_USB_CLIENT_H_
 #define DEBUGROUTER_NATIVE_SOCKET_USB_CLIENT_H_
 
+#include "debug_router/native/base/socket_guard.h"
 #include "debug_router/native/socket/blocking_queue.h"
 #include "debug_router/native/socket/count_down_latch.h"
 #include "debug_router/native/socket/socket_server_type.h"
@@ -41,13 +42,12 @@ class UsbClient : public std::enable_shared_from_this<UsbClient> {
   void DisconnectInternal();
   void SendInternal(const std::string &message);
 
-  void StartReader(SocketType socket_fd_);
-  void StartWriter(SocketType socket_fd_);
-  void StartMessageDispatcher(SocketType socket_fd_);
-  void ReadMessage(SocketType socket_fd_);
+  void StartReader();
+  void StartWriter();
+  void StartMessageDispatcher();
+  void ReadMessage();
   void MessageDispatcher();
-  void WriteMessage(SocketType socket_fd_);
-  void HandleFirstFrame(SocketType socket_fd);
+  void WriteMessage();
 
   bool Read(SocketType socket_fd_, char *buffer, uint32_t read_size);
   bool ReadAndCheckMessageHeader(char *header, SocketType socket_fd_);
@@ -82,24 +82,19 @@ class UsbClient : public std::enable_shared_from_this<UsbClient> {
    */
   static void WrapHeader(const std::string &message, std::string &result);
 
-  // work threads
-  static void MessageDispatcherFunc(std::shared_ptr<UsbClient> client);
-  static void ReadThreadFunc(std::shared_ptr<UsbClient> client,
-                             SocketType socket_fd_);
-  static void WriteThreadFunc(std::shared_ptr<UsbClient> client,
-                              SocketType socket_fd_);
-
  private:
   BlockingQueue<std::string> incoming_message_queue_;
   BlockingQueue<std::string> outgoing_message_queue_;
 
   base::WorkThreadExecutor work_thread_;
+  base::WorkThreadExecutor read_thread_;
+  base::WorkThreadExecutor write_thread_;
+  base::WorkThreadExecutor dispatch_thread_;
   std::shared_ptr<UsbClientListener> listener_;
   USBConnectStatus connect_status_ = USBConnectStatus::DISCONNECTED;
   std::unique_ptr<CountDownLatch> latch_;
 
-  // TODO(popoaichuiniu) use unique_fd to manage socket
-  volatile SocketType socket_fd_ = kInvalidSocket;
+  base::SocketGuard socket_guard_;
   // mutex for close socket_fd_
   std::mutex mutex_;
 };
