[BugFix]Fix all the crash that caused by thread processing

1. fix do_read crash caused by multithreaded operation of the socket_guard_.
2. fix crash that run() thread can not be released by the destructor when work_thread_executor destructor.
3. add worker->get_id() to determine whether the current waiting thread is this thread, if so, detach.

Author: RovicnowYoyi

debug_router/native/net/websocket_task.cc

@@ -91,10 +91,14 @@ std::string decodeURIComponent(std::string url) {
 WebSocketTask::WebSocketTask(
     std::shared_ptr<core::MessageTransceiver> transceiver,
     const std::string &url)
-    : transceiver_(transceiver), url_(url) {
+    : transceiver_(transceiver),
+      url_(url),
+      socket_guard_(std::make_unique<base::SocketGuard>(kInvalidSocket)) {
   submit([this]() { start(); });
 }
 
+WebSocketTask::~WebSocketTask() { shutdown(); }
+
 void WebSocketTask::SendInternal(const std::string &data) {
   const char *buf = data.data();
   size_t payloadLen = data.size();

debug_router/native/net/websocket_task.h

@@ -16,6 +16,7 @@ class WebSocketTask : public base::WorkThreadExecutor {
  public:
   WebSocketTask(std::shared_ptr<core::MessageTransceiver> transceiver,
                 const std::string &url);
+  virtual ~WebSocketTask() override;
 
   void SendInternal(const std::string &data);
 

debug_router/native/socket/socket_server_api.cc

@@ -40,6 +40,9 @@ void SocketServer::HandleOnOpenStatus(std::shared_ptr<UsbClient> client,
   thread::DebugRouterExecutor::GetInstance().Post([=]() {
     std::shared_ptr<UsbClient> old_client_ = usb_client_;
     LOGI("SocketServerApi OnOpen: replace old client.");
+    if (old_client_) {
+      old_client_->Stop();
+    }
     usb_client_ = client;
     if (auto listener = listener_.lock()) {
       listener->OnStatusChanged(kConnected, code, reason);
@@ -68,6 +71,7 @@ void SocketServer::HandleOnCloseStatus(std::shared_ptr<UsbClient> client,
       LOGI("SocketServerApi OnMessage: client is null or not match.");
       return;
     }
+    usb_client_->Stop();
     usb_client_ = nullptr;
     if (auto listener = listener_.lock()) {
       listener->OnStatusChanged(status, code, reason);
@@ -83,6 +87,7 @@ void SocketServer::HandleOnErrorStatus(std::shared_ptr<UsbClient> client,
       LOGI("SocketServerApi OnMessage: client is null or not match.");
       return;
     }
+    usb_client_->Stop();
     usb_client_ = nullptr;
     if (auto listener = listener_.lock()) {
       listener->OnStatusChanged(status, code, reason);

debug_router/native/socket/usb_client.cc

@@ -367,6 +367,8 @@ bool UsbClient::Send(const std::string &message) {
   return true;
 }
 
+void UsbClient::Stop() { work_thread_.shutdown(); }
+
 void UsbClient::SendInternal(const std::string &message) {
   LOGI("UsbClient: SendInternal.");
   if (connect_status_ != USBConnectStatus::CONNECTED) {

debug_router/native/socket/usb_client.h

@@ -28,6 +28,8 @@ class UsbClient : public std::enable_shared_from_this<UsbClient> {
   // true means the message are added to message queue
   bool Send(const std::string &message);
 
+  void Stop();
+
   explicit UsbClient(SocketType socket_fd);
   ~UsbClient();
 

debug_router/native/socket/work_thread_executor.cc

@@ -41,7 +41,11 @@ void WorkThreadExecutor::shutdown() {
 
   if (worker && worker->joinable()) {
     try {
-      worker->join();
+      if (worker->get_id() != std::this_thread::get_id()) {
+        worker->join();
+      } else {
+        worker->detach();
+      }
     } catch (const std::exception& e) {
       LOGE("WorkThreadExecutor::shutdown worker->detach() failed, "
            << e.what());

debug_router/native/socket/work_thread_executor.h

@@ -17,7 +17,7 @@ namespace base {
 class WorkThreadExecutor {
  public:
   WorkThreadExecutor();
-  ~WorkThreadExecutor();
+  virtual ~WorkThreadExecutor();
 
   void submit(std::function<void()> task);
   void shutdown();