[BugFix] Fix crash of worker_thread_executor

  1. start thread after constructor
  2. add weak_ptr to verify whether the class is been destroyed.

Author: RovicnowYoyi

debug_router/native/net/websocket_client.cc

@@ -36,7 +36,7 @@ WebSocketClient::WebSocketClient() {}
 
 WebSocketClient::~WebSocketClient() { DisconnectInternal(); }
 
-void WebSocketClient::Init() {}
+void WebSocketClient::Init() { work_thread_.init(); }
 
 bool WebSocketClient::Connect(const std::string &url) {
   LOGI("WebSocketClient::Connect");
@@ -51,6 +51,8 @@ bool WebSocketClient::Connect(const std::string &url) {
 void WebSocketClient::ConnectInternal(const std::string &url) {
   LOGI("WebSocketClient::ConnectInternal: use " << url << " to connect.");
   current_task_ = std::make_unique<WebSocketTask>(shared_from_this(), url);
+  current_task_->init();
+  current_task_->Start();
 }
 
 void WebSocketClient::Disconnect() {

debug_router/native/net/websocket_task.cc

@@ -44,9 +44,7 @@ WebSocketTask::WebSocketTask(
     const std::string &url)
     : transceiver_(transceiver),
       url_(url),
-      socket_guard_(std::make_unique<base::SocketGuard>(kInvalidSocket)) {
-  submit([this]() { start(); });
-}
+      socket_guard_(std::make_unique<base::SocketGuard>(kInvalidSocket)) {}
 
 WebSocketTask::~WebSocketTask() { shutdown(); }
 
@@ -98,7 +96,11 @@ void WebSocketTask::SendInternal(const std::string &data) {
   LOGI("send: prefix_len and buf success.");
 }
 
-void WebSocketTask::start() {
+void WebSocketTask::Start() {
+  submit([this]() { StartInternal(); });
+}
+
+void WebSocketTask::StartInternal() {
   if (!do_connect()) {
     onFailure();
     return;

debug_router/native/net/websocket_task.h

@@ -19,10 +19,11 @@ class WebSocketTask : public base::WorkThreadExecutor {
   virtual ~WebSocketTask() override;
 
   void Stop();
+  void Start();
   void SendInternal(const std::string &data);
 
  private:
-  void start();
+  void StartInternal();
 
   bool do_connect();
 

debug_router/native/socket/posix/socket_server_posix.cc

@@ -94,10 +94,11 @@ void SocketServerPosix::Start() {
     NotifyInit(GetErrorMessage(), "accept socket error");
     return;
   }
-  auto current_usb_client_ = std::make_shared<UsbClient>(accept_socket_fd);
+  auto temp_usb_client = std::make_shared<UsbClient>(accept_socket_fd);
   std::shared_ptr<ClientListener> listener =
       std::make_shared<ClientListener>(shared_from_this());
-  current_usb_client_->StartUp(listener);
+  temp_usb_client->Init();
+  temp_usb_client->StartUp(listener);
 }
 
 void SocketServerPosix::CloseSocket(int socket_fd) {

debug_router/native/socket/usb_client.cc

@@ -55,6 +55,8 @@ void UsbClient::SetConnectStatus(USBConnectStatus status) {
   });
 }
 
+void UsbClient::Init() { work_thread_.init(); }
+
 void UsbClient::StartUp(const std::shared_ptr<UsbClientListener> &listener) {
   LOGI("UsbClient: StartUp.");
   work_thread_.submit([client_ptr = shared_from_this(), listener]() {

debug_router/native/socket/usb_client.h

@@ -23,6 +23,7 @@ static const char *kMessageQuit = "quit";
 // Client of socket_server
 class UsbClient : public std::enable_shared_from_this<UsbClient> {
  public:
+  void Init();
   // below three functions work only on one work thread
   void StartUp(const std::shared_ptr<UsbClientListener> &listener);
   // true means the message are added to message queue

debug_router/native/socket/work_thread_executor.cc

@@ -9,8 +9,14 @@
 namespace debugrouter {
 namespace base {
 
-WorkThreadExecutor::WorkThreadExecutor() : is_shut_down(false) {
-  worker = std::make_unique<std::thread>([this]() { run(); });
+WorkThreadExecutor::WorkThreadExecutor()
+    : is_shut_down(false), alive_flag(std::make_shared<bool>(true)) {}
+
+void WorkThreadExecutor::init() {
+  std::lock_guard<std::mutex> lock(task_mtx);
+  if (!worker) {
+    worker = std::make_unique<std::thread>([this]() { run(); });
+  }
 }
 
 WorkThreadExecutor::~WorkThreadExecutor() { shutdown(); }
@@ -62,7 +68,15 @@ void WorkThreadExecutor::shutdown() {
 }
 
 void WorkThreadExecutor::run() {
-  while (!is_shut_down) {
+  std::weak_ptr<bool> weak_flag = alive_flag;
+  while (true) {
+    auto flag = weak_flag.lock();
+    if (!flag) {
+      break;
+    }
+    if (is_shut_down) {
+      break;
+    }
     std::unique_lock<std::mutex> lock(task_mtx);
     cond.wait(lock, [this] { return !tasks.empty() || is_shut_down; });
     if (is_shut_down) {

debug_router/native/socket/work_thread_executor.h

@@ -19,6 +19,7 @@ class WorkThreadExecutor {
   WorkThreadExecutor();
   virtual ~WorkThreadExecutor();
 
+  void init();
   void submit(std::function<void()> task);
   void shutdown();
 
@@ -30,6 +31,7 @@ class WorkThreadExecutor {
   std::queue<std::function<void()>> tasks;
   std::mutex task_mtx;
   std::condition_variable cond;
+  std::shared_ptr<bool> alive_flag;
 };
 
 }  // namespace base