[[415983]]

之前的文章分析了Node.js Inspector的使用和原理，并粗略地分析了其源碼，因為Node.js Inspector的實現非常復雜，邏輯又非常繞，所以本文打算更深入、更通俗地講解Node.js Inspector的實現。

當我們以以下方式執行我們的應用時

node inspect app.js 

1 初始化

Node.js在啟動的過程中，就會初始化Inspector相關的邏輯。

inspector_agent_ = std::make_unique<inspector::Agent>(this); 

Agent是負責和V8 Inspector通信的對象。創建完后接著執行env->InitializeInspector({})啟動Agent。

inspector_agent_->Start(...); 

Start繼續執行Agent::StartIoThread。

bool Agent::StartIoThread() { 
  io_ = InspectorIo::Start(client_->getThreadHandle(), ...); 
  return true; 
} 

StartIoThread中的client_->getThreadHandle()是重要的邏輯，我們先來分析該函數。

std::shared_ptr<MainThreadHandle> getThreadHandle() { 
    if (!interface_) { 
      interface_ = std::make_shared<MainThreadInterface>(env_->inspector_agent(), ...); 
    } 
    return interface_->GetHandle(); 
} 

getThreadHandle首先創建來一個MainThreadInterface對象，接著又調用了他的GetHandle方法，我們看一下該方法的邏輯。

std::shared_ptr<MainThreadHandle> MainThreadInterface::GetHandle() { 
  if (handle_ == nullptr) 
    handle_ = std::make_shared<MainThreadHandle>(this); 
  return handle_; 
} 

GetHandlei了創建了一個MainThreadHandle對象，最終結構如下所示。

分析完后我們繼續看Agent::StartIoThread中InspectorIo::Start的邏輯。

std::unique_ptr<InspectorIo> InspectorIo::Start(std::shared_ptr<MainThreadHandle> main_thread, ...) { 
  auto io = std::unique_ptr<InspectorIo>(new InspectorIo(main_thread, ...)); 
  return io; 
} 

InspectorIo::Star里新建了一個InspectorIo對象，我們看看InspectorIo構造函數的邏輯。

InspectorIo::InspectorIo(std::shared_ptr<MainThreadHandle> main_thread, ...) 
    :  
    // 初始化main_thread_ 
    main_thread_(main_thread)) { 
  // 新建一個子線程，子線程中執行InspectorIo::ThreadMain 
  uv_thread_create(&thread_, InspectorIo::ThreadMain, this); 
} 

這時候結構如下。

Inspector在子線程里啟動的原因主要有兩個。

1 如果在主線程里運行，那么當我們斷點調試的時候，Node.js主線程就會被停住，也就無法處理客戶端發過來的調試指令。

2 如果主線程陷入死循環，我們就無法實時抓取進程的profile數據來分析原因。接著繼續看一下子線程里執行InspectorIo::ThreadMain的邏輯。

void InspectorIo::ThreadMain(void* io) { 
  static_cast<InspectorIo*>(io)->ThreadMain(); 
} 
 
void InspectorIo::ThreadMain() { 
  uv_loop_t loop; 
  loop.data = nullptr; 
  // 在子線程開啟一個新的事件循環 
  int err = uv_loop_init(&loop); 
  std::shared_ptr<RequestQueueData> queue(new RequestQueueData(&loop), ...); 
  // 新建一個delegate，用于處理請求 
  std::unique_ptr<InspectorIoDelegate> delegate( 
    new InspectorIoDelegate(queue, main_thread_, ...) 
  ); 
  InspectorSocketServer server(std::move(delegate), ...); 
  server.Start() 
  uv_run(&loop, UV_RUN_DEFAULT); 
} 

ThreadMain里主要三個邏輯

1 創建一個delegate對象，該對象是核心的對象，后面我們會看到有什么作用。

2 創建一個服務器并啟動。

3 開啟事件循環。接下來看一下服務器的邏輯，首先看一下創建服務器的邏輯。

InspectorSocketServer::InspectorSocketServer(std::unique_ptr<SocketServerDelegate> delegate, ...) 
    :  
      // 保存delegate 
      delegate_(std::move(delegate)), 
      // 初始化sessionId 
      next_session_id_(0) { 
  // 設置delegate的server為當前服務器 
  delegate_->AssignServer(this); 
} 

執行完后形成以下結構。

接著我們看啟動服務器的邏輯。

bool InspectorSocketServer::Start() { 
  // DNS解析,比如輸入的是localhost 
  struct addrinfo hints; 
  memset(&hints, 0, sizeof(hints)); 
  hints.ai_flags = AI_NUMERICSERV; 
  hints.ai_socktype = SOCK_STREAM; 
  uv_getaddrinfo_t req; 
  const std::string port_string = std::to_string(port_); 
  uv_getaddrinfo(loop_, &req, nullptr, host_.c_str(), 
                           port_string.c_str(), &hints); 
  // 監聽解析到的ip列表                  
  for (addrinfo* address = req.addrinfo;  
       address != nullptr; 
       address = address->ai_next) { 
 
    auto server_socket = ServerSocketPtr(new ServerSocket(this)); 
    err = server_socket->Listen(address->ai_addr, loop_); 
    if (err == 0) 
      server_sockets_.push_back(std::move(server_socket)); 
 
  } 
 
  return true; 
} 

首先根據參數做一個DNS解析，然后根據拿到的ip列表(通常是一個)，創建對應個數的ServerSocket對象，并執行他的Listen方法。ServerSocket表示一個監聽socket。看一下ServerSocket的構造函數。

ServerSocket(InspectorSocketServer* server) 
            : tcp_socket_(uv_tcp_t()), server_(server) {} 

執行完后結構如下。

接著看一下ServerSocket的Listen方法。

int ServerSocket::Listen(sockaddr* addr, uv_loop_t* loop) { 
  uv_tcp_t* server = &tcp_socket_; 
  uv_tcp_init(loop, server) 
  uv_tcp_bind(server, addr, 0); 
  uv_listen(reinterpret_cast<uv_stream_t*>(server),  
                    511, 
                    ServerSocket::SocketConnectedCallback); 
} 

Listen調用Libuv的接口完成服務器的啟動。至此，Inspector提供的Weboscket服務器啟動了。

2 處理連接

從剛才分析中可以看到，當有連接到來時執行回調ServerSocket::SocketConnectedCallback。

void ServerSocket::SocketConnectedCallback(uv_stream_t* tcp_socket, 
                                           int status) { 
  if (status == 0) { 
    // 根據Libuv handle找到對應的ServerSocket對象 
    ServerSocket* server_socket = ServerSocket::FromTcpSocket(tcp_socket); 
    // Socket對象的server_字段保存了所在的InspectorSocketServer 
    server_socket->server_->Accept(server_socket->port_, tcp_socket); 
  } 
} 

接著看InspectorSocketServer的Accept是如何處理連接的。

void InspectorSocketServer::Accept(int server_port, 
                                   uv_stream_t* server_socket) { 
 
  std::unique_ptr<SocketSession> session( 
      new SocketSession(this, next_session_id_++, server_port) 
  ); 
 
  InspectorSocket::DelegatePointer delegate = 
      InspectorSocket::DelegatePointer( 
          new SocketSession::Delegate(this, session->id()) 
      ); 
 
  InspectorSocket::Pointer inspector = 
      InspectorSocket::Accept(server_socket, std::move(delegate)); 
 
  if (inspector) { 
    session->Own(std::move(inspector)); 
    connected_sessions_[session->id()].second = std::move(session); 
  } 
} 

Accept的首先創建里一個SocketSession和SocketSession::Delegate對象。然后調用InspectorSocket::Accept，從代碼中可以看到InspectorSocket::Accept會返回一個InspectorSocket對象。InspectorSocket是對通信socket的封裝(和客戶端通信的socket，區別于服務器的監聽socket)。然后記錄session對象對應的InspectorSocket對象，同時記錄sessionId和session的映射關系。結構如下圖所示。

接著看一下InspectorSocket::Accept返回InspectorSocket的邏輯。

InspectorSocket::Pointer InspectorSocket::Accept(uv_stream_t* server, 
                                                 DelegatePointer delegate) { 
  auto tcp = TcpHolder::Accept(server, std::move(delegate)); 
  InspectorSocket* inspector = new InspectorSocket(); 
  inspector->SwitchProtocol(new HttpHandler(inspector, std::move(tcp))); 
  return InspectorSocket::Pointer(inspector); 
} 

InspectorSocket::Accept的代碼不多，但是邏輯還是挺多的。

TcpHolder::Pointer TcpHolder::Accept( 
    uv_stream_t* server, 
    InspectorSocket::DelegatePointer delegate) { 
  // 新建一個TcpHolder對象，TcpHolder是對uv_tcp_t和delegate的封裝 
  TcpHolder* result = new TcpHolder(std::move(delegate)); 
  // 拿到TcpHolder對象的uv_tcp_t結構體 
  uv_stream_t* tcp = reinterpret_cast<uv_stream_t*>(&result->tcp_); 
  // 初始化 
  int err = uv_tcp_init(server->loop, &result->tcp_); 
  // 摘取一個TCP連接對應的fd保存到TcpHolder的uv_tcp_t結構體中（即第二個參數的tcp字段） 
  uv_accept(server, tcp); 
  // 注冊等待可讀事件，有數據時執行OnDataReceivedCb回調 
  uv_read_start(tcp, allocate_buffer, OnDataReceivedCb); 
  return TcpHolder::Pointer(result); 
} 

2 新建一個HttpHandler對象。

explicit HttpHandler(InspectorSocket* inspector, TcpHolder::Pointer tcp) 
                     : ProtocolHandler(inspector, std::move(tcp)){ 
 
  llhttp_init(&parser_, HTTP_REQUEST, &parser_settings); 
  llhttp_settings_init(&parser_settings); 
  parser_settings.on_header_field = OnHeaderField; 
  parser_settings.on_header_value = OnHeaderValue; 
  parser_settings.on_message_complete = OnMessageComplete; 
  parser_settings.on_url = OnPath; 
} 
 
ProtocolHandler::ProtocolHandler(InspectorSocket* inspector, 
                                 TcpHolder::Pointer tcp) 
                                 : inspector_(inspector), tcp_(std::move(tcp)) { 
  // 設置TCP數據的handler，TCP是只負責傳輸，數據的解析交給handler處理                                
  tcp_->SetHandler(this); 
} 

HttpHandler是對uv_tcp_t的封裝，主要通過HTTP解析器llhttp對HTTP協議進行解析。

3 調用inspector->SwitchProtocol()切換當前協議為HTTP，建立TCP連接后，首先要經過一個HTTP請求從HTTP協議升級到WebSocket協議，升級成功后就使用Websocket協議進行通信。我們看一下這時候的結構圖。

至此，就完成了連接處理的分析。

3 協議升級

完成了TCP連接的處理后，接下來要完成協議升級，因為Inspector是通過WebSocket協議和客戶端通信的，所以需要通過一個HTTP請求來完成HTTP到WebSocekt協議的升級。從剛才的分析中看當有數據到來時會執行OnDataReceivedCb回調。

void TcpHolder::OnDataReceivedCb(uv_stream_t* tcp, ssize_t nread, 
                                 const uv_buf_t* buf) { 
  TcpHolder* holder = From(tcp); 
  holder->ReclaimUvBuf(buf, nread); 
  // 調用handler的onData，目前handler是HTTP協議 
  holder->handler_->OnData(&holder->buffer); 
} 

TCP層收到數據后交給應用層解析，直接調用上層的OnData回調。

void OnData(std::vector<char>* data) override { 
    // 解析HTTP協議 
    llhttp_execute(&parser_, data->data(), data->size()); 
    // 解析完并且是升級協議的請求則調用delegate的回調OnSocketUpgrade 
    delegate()->OnSocketUpgrade(event.host, event.path, event.ws_key); 
} 

OnData可能會被多次回調，并通過llhttp_execute解析收到的HTTP報文，當發現是一個協議升級的請求后，就調用OnSocketUpgrade回調。delegate是TCP層保存的SocketSession::Delegate對象。來看一下該對象的OnSocketUpgrade方法。

void SocketSession::Delegate::OnSocketUpgrade(const std::string& host, 
                                              const std::string& path, 
                                              const std::string& ws_key) { 
  std::string id = path.empty() ? path : path.substr(1); 
  server_->SessionStarted(session_id_, id, ws_key); 
} 

OnSocketUpgrade又調用來server_(InspectorSocketServer對象)的SessionStarted。

void InspectorSocketServer::SessionStarted(int session_id, 
                                           const std::string& id, 
                                           const std::string& ws_key) { 
  // 找到對應的session對象                                            
  SocketSession* session = Session(session_id); 
  connected_sessions_[session_id].first = id; 
  session->Accept(ws_key); 
  delegate_->StartSession(session_id, id); 
} 

首先通過session_id找到建立TCP連接時分配的SocketSession對象。

1 執行session->Accept(ws_key);回復客戶端同意協議升級。

void Accept(const std::string& ws_key) { 
  ws_socket_->AcceptUpgrade(ws_key); 
} 

從結構圖我們可以看到ws_socket_是一個InspectorSocket對象。

void AcceptUpgrade(const std::string& accept_key) override { 
    char accept_string[ACCEPT_KEY_LENGTH]; 
    generate_accept_string(accept_key, &accept_string); 
    const char accept_ws_prefix[] = "HTTP/1.1 101 Switching Protocols\r\n" 
                                    "Upgrade: websocket\r\n" 
                                    "Connection: Upgrade\r\n" 
                                    "Sec-WebSocket-Accept: "; 
    const char accept_ws_suffix[] = "\r\n\r\n"; 
    std::vector<char> reply(accept_ws_prefix, 
                            accept_ws_prefix + sizeof(accept_ws_prefix) - 1); 
    reply.insert(reply.end(), accept_string, 
                 accept_string + sizeof(accept_string)); 
    reply.insert(reply.end(), accept_ws_suffix, 
                 accept_ws_suffix + sizeof(accept_ws_suffix) - 1); 
    // 回復101給客戶端              
    WriteRaw(reply, WriteRequest::Cleanup); 
    // 切換handler為WebSocket handler 
    inspector_->SwitchProtocol(new WsHandler(inspector_, std::move(tcp_))); 
} 

AcceptUpgradeh首先回復客戶端101表示同意升級道WebSocket協議，然后切換數據處理器為WsHandler，即后續的數據按照WebSocket協議處理。

2 執行delegate_->StartSession(session_id, id)建立和V8 Inspector的會話。delegate_是InspectorIoDelegate對象。

void InspectorIoDelegate::StartSession(int session_id, 
                                       const std::string& target_id) { 
  auto session = main_thread_->Connect( 
      std::unique_ptr<InspectorSessionDelegate>( 
          new IoSessionDelegate(request_queue_->handle(), session_id) 
      ),  
      true); 
  if (session) { 
    sessions_[session_id] = std::move(session); 
    fprintf(stderr, "Debugger attached.\n"); 
  } 
} 

首先通過main_thread_->Connect拿到一個session，并在InspectorIoDelegate中記錄映射關系。結構圖如下。

接下來看一下main_thread_->Connect的邏輯(main_thread_是MainThreadHandle對象)。

std::unique_ptr<InspectorSession> MainThreadHandle::Connect( 
    std::unique_ptr<InspectorSessionDelegate> delegate, 
    bool prevent_shutdown) { 
 
  return std::unique_ptr<InspectorSession>( 
      new CrossThreadInspectorSession(++next_session_id_, 
                                      shared_from_this(), 
                                      std::move(delegate), 
                                      prevent_shutdown)); 
} 

Connect函數新建了一個CrossThreadInspectorSession對象。

CrossThreadInspectorSession( 
      int id, 
      std::shared_ptr<MainThreadHandle> thread, 
      std::unique_ptr<InspectorSessionDelegate> delegate, 
      bool prevent_shutdown) 
      // 創建一個MainThreadSessionState對象 
      : state_(thread, std::bind(MainThreadSessionState::Create, 
                                 std::placeholders::_1, 
                                 prevent_shutdown)) { 
    // 執行MainThreadSessionState::Connect                              
    state_.Call(&MainThreadSessionState::Connect, std::move(delegate)); 
  } 

繼續看MainThreadSessionState::Connect。

void Connect(std::unique_ptr<InspectorSessionDelegate> delegate) { 
    Agent* agent = thread_->inspector_agent(); 
    session_ = agent->Connect(std::move(delegate), prevent_shutdown_); 
} 

繼續調agent->Connect。

std::unique_ptr<InspectorSession> Agent::Connect( 
    std::unique_ptr<InspectorSessionDelegate> delegate, 
    bool prevent_shutdown) { 
 
  int session_id = client_->connectFrontend(std::move(delegate), 
                                            prevent_shutdown); 
  return std::unique_ptr<InspectorSession>( 
      new SameThreadInspectorSession(session_id, client_)); 
} 

繼續調connectFrontend

int connectFrontend(std::unique_ptr<InspectorSessionDelegate> delegate, 
                      bool prevent_shutdown) { 
    int session_id = next_session_id_++; 
    channels_[session_id] = std::make_unique<ChannelImpl>(env_, 
                                                          client_, 
                                                          getWorkerManager(), 
                                                          std::move(delegate), 
                                                          getThreadHandle(), 
                                                          prevent_shutdown); 
    return session_id; 
} 

connectFrontend創建了一個ChannelImpl并且在channels_中保存了映射關系。看看ChannelImpl的構造函數。

explicit ChannelImpl(Environment* env, 
                     const std::unique_ptr<V8Inspector>& inspector, 
                     std::unique_ptr<InspectorSessionDelegate> delegate, ...) 
      : delegate_(std::move(delegate)) { 
 
    session_ = inspector->connect(CONTEXT_GROUP_ID, this, StringView()); 
} 

ChannelImpl調用inspector->connect建立了一個和V8 Inspector的會話。結構圖大致如下。

4 客戶端到V8 Inspector的數據處理

TCP連接建立了，協議升級也完成了，接下來就可以開始處理業務數據。從前面的分析中我們已經知道數據到來時會執行TcpHoldler的handler_->OnData回調。因為已經完成了協議升級，所以這時候的handler變成了WeSocket handler。

void OnData(std::vector<char>* data) override { 
    // 1. Parse. 
    int processed = 0; 
    do { 
      processed = ParseWsFrames(*data); 
      // 2. Fix the data size & length 
      if (processed > 0) { 
        remove_from_beginning(data, processed); 
      } 
    } while (processed > 0 && !data->empty()); 
} 

OnData通過ParseWsFrames解析WebSocket協議。

int ParseWsFrames(const std::vector<char>& buffer) { 
    int bytes_consumed = 0; 
    std::vector<char> output; 
    bool compressed = false; 
    // 解析WebSocket協議 
    ws_decode_result r =  decode_frame_hybi17(buffer, 
                                              true /* client_frame */, 
                                              &bytes_consumed, &output, 
                                              &compressed); 
    // 執行delegate的回調                                         
    delegate()->OnWsFrame(output); 
    return bytes_consumed; 
} 

前面已經分析過delegate是TcpHoldler的delegate，即SocketSession::Delegate對象。

void SocketSession::Delegate::OnWsFrame(const std::vector<char>& data) { 
  server_->MessageReceived(session_id_, 
                           std::string(data.data(),  
                           data.size())); 
} 

繼續回調server_->MessageReceived。從結構圖可以看到server_是InspectorSocketServer對象。

void MessageReceived(int session_id, const std::string& message) { 
  delegate_->MessageReceived(session_id, message); 
} 

繼續回調delegate_->MessageReceived。InspectorSocketServer的delegate_是InspectorIoDelegate對象。

void InspectorIoDelegate::MessageReceived(int session_id, 
                                          const std::string& message) { 
  auto session = sessions_.find(session_id); 
  if (session != sessions_.end()) 
    session->second->Dispatch(Utf8ToStringView(message)->string()); 
} 

首先通過session_id找到對應的session。session是一個CrossThreadInspectorSession對象。看看他的Dispatch方法。

void Dispatch(const StringView& message) override { 
    state_.Call(&MainThreadSessionState::Dispatch, 
                StringBuffer::create(message)); 
} 

執行MainThreadSessionState::Dispatch。

void Dispatch(std::unique_ptr<StringBuffer> message) { 
  session_->Dispatch(message->string()); 
} 

session_是SameThreadInspectorSession對象。

void SameThreadInspectorSession::Dispatch( 
    const v8_inspector::StringView& message) { 
  auto client = client_.lock(); 
  if (client) 
    client->dispatchMessageFromFrontend(session_id_, message); 
} 

繼續調client->dispatchMessageFromFrontend。

void dispatchMessageFromFrontend(int session_id, const StringView& message) { 
   channels_[session_id]->dispatchProtocolMessage(message); 
} 

通過session_id找到對應的ChannelImpl，繼續調ChannelImpl的dispatchProtocolMessage。

voiddispatchProtocolMessage(const StringView& message) { 
   session_->dispatchProtocolMessage(message); 
} 

最終調用和V8 Inspector的會話對象把數據發送給V8。至此客戶端到V8 Inspector的通信過程就完成了。

5 V8 Inspector到客戶端的數據處理

接著看從V8 inspector到客戶端的數據傳遞邏輯。V8 inspector是通過channel的sendResponse函數傳遞給客戶端的。

void sendResponse( 
      int callId, 
      std::unique_ptr<v8_inspector::StringBuffer> message) override { 
 
    sendMessageToFrontend(message->string()); 
  } 
 
 void sendMessageToFrontend(const StringView& message) { 
    delegate_->SendMessageToFrontend(message); 
 } 

delegate_是IoSessionDelegate對象。

void SendMessageToFrontend(const v8_inspector::StringView& message) override { 
    request_queue_->Post(id_, TransportAction::kSendMessage, 
                         StringBuffer::create(message)); 
  } 

request_queue_是RequestQueueData對象。

void Post(int session_id, 
            TransportAction action, 
            std::unique_ptr<StringBuffer> message) { 
 
    Mutex::ScopedLock scoped_lock(state_lock_); 
    bool notify = messages_.empty(); 
    messages_.emplace_back(action, session_id, std::move(message)); 
    if (notify) { 
      CHECK_EQ(0, uv_async_send(&async_)); 
      incoming_message_cond_.Broadcast(scoped_lock); 
    } 
  } 

Post首先把消息入隊，然后通過異步的方式通知async_接著看async_的處理函數(在子線程的事件循環里執行)。

uv_async_init(loop, &async_, [](uv_async_t* async) { 
   // 拿到async對應的上下文 
   RequestQueueData* wrapper = node::ContainerOf(&RequestQueueData::async_, async); 
   // 執行RequestQueueData的DoDispatch 
   wrapper->DoDispatch();});void DoDispatch() { 
    for (const auto& request : GetMessages()) { 
      request.Dispatch(server_); 
    } 
  } 

request是RequestToServer對象。

void Dispatch(InspectorSocketServer* server) const { 
    switch (action_) { 
      case TransportAction::kSendMessage: 
        server->Send( 
            session_id_, 
            protocol::StringUtil::StringViewToUtf8(message_->string())); 
        break; 
    } 
  } 

接著看InspectorSocketServer的Send。

void InspectorSocketServer::Send(int session_id, const std::string& message) { 
  SocketSession* session = Session(session_id); 
  if (session != nullptr) { 
    session->Send(message); 
  } 
} 

session代表可客戶端的一個連接。

void SocketSession::Send(const std::string& message) { 
  ws_socket_->Write(message.data(), message.length()); 
} 

接著調用WebSocket handler的Write。

void Write(const std::vector<char> data) override { 
    std::vector<char> output = encode_frame_hybi17(data); 
    WriteRaw(output, WriteRequest::Cleanup); 
  } 

WriteRaw是基類ProtocolHandler實現的。

int ProtocolHandler::WriteRaw(const std::vector<char>& buffer, 
                              uv_write_cb write_cb) { 
  return tcp_->WriteRaw(buffer, write_cb); 
} 

最終是通過TCP連接返回給客戶端。

int TcpHolder::WriteRaw(const std::vector<char>& buffer, uv_write_cb write_cb) { 
  // Freed in write_request_cleanup 
  WriteRequest* wr = new WriteRequest(handler_, buffer); 
  uv_stream_t* stream = reinterpret_cast<uv_stream_t*>(&tcp_); 
  int err = uv_write(&wr->req, stream, &wr->buf, 1, write_cb); 
  if (err < 0) 
    delete wr; 
  return err < 0; 
} 

新建一個寫請求，socket可寫的時候發送數據給客戶端。

 

后記：Node.js Inspector的原理雖然不復雜的，但是實現實在太繞了。