inspector: fix process._debugEnd() for inspector

This change ensures that the WebSocket server can be stopped
(and restarted if needed) buy calling process._debugEnd.

PR-URL: https://github.com/nodejs/node/pull/12777
Fixes: https://github.com/nodejs/node/issues/12559
Reviewed-By: Sam Roberts <vieuxtech@gmail.com>
Reviewed-By: Refael Ackermann <refack@gmail.com>

src/inspector_agent.cc

@@ -38,6 +38,18 @@ using v8_inspector::V8Inspector;
 static uv_sem_t inspector_io_thread_semaphore;
 static uv_async_t start_inspector_thread_async;
 
+class StartIoTask : public v8::Task {
+ public:
+  explicit StartIoTask(Agent* agent) : agent(agent) {}
+
+  void Run() override {
+    agent->StartIoThread(false);
+  }
+
+ private:
+  Agent* agent;
+};
+
 std::unique_ptr<StringBuffer> ToProtocolString(Isolate* isolate,
                                                Local<Value> value) {
   TwoByteValue buffer(isolate, value);
@@ -46,9 +58,14 @@ std::unique_ptr<StringBuffer> ToProtocolString(Isolate* isolate,
 
 // Called from the main thread.
 void StartInspectorIoThreadAsyncCallback(uv_async_t* handle) {
-  static_cast<Agent*>(handle->data)->StartIoThread();
+  static_cast<Agent*>(handle->data)->StartIoThread(false);
+}
+
+void StartIoCallback(Isolate* isolate, void* agent) {
+  static_cast<Agent*>(agent)->StartIoThread(false);
 }
 
+
 #ifdef __POSIX__
 static void EnableInspectorIOThreadSignalHandler(int signo) {
   uv_sem_post(&inspector_io_thread_semaphore);
@@ -57,7 +74,9 @@ static void EnableInspectorIOThreadSignalHandler(int signo) {
 inline void* InspectorIoThreadSignalThreadMain(void* unused) {
   for (;;) {
     uv_sem_wait(&inspector_io_thread_semaphore);
-    uv_async_send(&start_inspector_thread_async);
+    Agent* agent = static_cast<Agent*>(start_inspector_thread_async.data);
+    if (agent != nullptr)
+      agent->RequestIoStart();
   }
   return nullptr;
 }
@@ -103,7 +122,9 @@ static int RegisterDebugSignalHandler() {
 
 #ifdef _WIN32
 DWORD WINAPI EnableDebugThreadProc(void* arg) {
-  uv_async_send(&start_inspector_thread_async);
+  Agent* agent = static_cast<Agent*>(start_inspector_thread_async.data);
+  if (agent != nullptr)
+    agent->RequestIoStart();
   return 0;
 }
 
@@ -387,21 +408,20 @@ bool Agent::Start(v8::Platform* platform, const char* path,
           new NodeInspectorClient(parent_env_, platform));
   inspector_->contextCreated(parent_env_->context(), "Node.js Main Context");
   platform_ = platform;
+  CHECK_EQ(0, uv_async_init(uv_default_loop(),
+                            &start_inspector_thread_async,
+                            StartInspectorIoThreadAsyncCallback));
+  start_inspector_thread_async.data = this;
+  uv_unref(reinterpret_cast<uv_handle_t*>(&start_inspector_thread_async));
+
+  RegisterDebugSignalHandler();
   if (options.inspector_enabled()) {
-    return StartIoThread();
-  } else {
-    CHECK_EQ(0, uv_async_init(uv_default_loop(),
-                              &start_inspector_thread_async,
-                              StartInspectorIoThreadAsyncCallback));
-    start_inspector_thread_async.data = this;
-    uv_unref(reinterpret_cast<uv_handle_t*>(&start_inspector_thread_async));
-
-    RegisterDebugSignalHandler();
-    return true;
+    return StartIoThread(options.wait_for_connect());
   }
+  return true;
 }
 
-bool Agent::StartIoThread() {
+bool Agent::StartIoThread(bool wait_for_connect) {
   if (io_ != nullptr)
     return true;
 
@@ -409,7 +429,8 @@ bool Agent::StartIoThread() {
 
   enabled_ = true;
   io_ = std::unique_ptr<InspectorIo>(
-      new InspectorIo(parent_env_, platform_, path_, debug_options_));
+      new InspectorIo(parent_env_, platform_, path_, debug_options_,
+                      wait_for_connect));
   if (!io_->Start()) {
     inspector_.reset();
     return false;
@@ -440,8 +461,10 @@ bool Agent::StartIoThread() {
 }
 
 void Agent::Stop() {
-  if (io_ != nullptr)
+  if (io_ != nullptr) {
     io_->Stop();
+    io_.reset();
+  }
 }
 
 void Agent::Connect(InspectorSessionDelegate* delegate) {
@@ -502,6 +525,18 @@ void Agent::InitJSBindings(Local<Object> target, Local<Value> unused,
   if (agent->debug_options_.wait_for_connect())
     env->SetMethod(target, "callAndPauseOnStart", CallAndPauseOnStart);
 }
+
+void Agent::RequestIoStart() {
+  // We need to attempt to interrupt V8 flow (in case Node is running
+  // continuous JS code) and to wake up libuv thread (in case Node is wating
+  // for IO events)
+  uv_async_send(&start_inspector_thread_async);
+  v8::Isolate* isolate = parent_env_->isolate();
+  platform_->CallOnForegroundThread(isolate, new StartIoTask(this));
+  isolate->RequestInterrupt(StartIoCallback, this);
+  uv_async_send(&start_inspector_thread_async);
+}
+
 }  // namespace inspector
 }  // namespace node
 

src/inspector_agent.h

@@ -51,7 +51,6 @@ class Agent {
 
   bool Start(v8::Platform* platform, const char* path,
              const DebugOptions& options);
-  bool StartIoThread();
   void Stop();
 
   bool IsStarted();
@@ -72,6 +71,13 @@ class Agent {
                              v8::Local<v8::Context> context,
                              void* priv);
 
+  bool StartIoThread(bool wait_for_connect);
+  InspectorIo* io() {
+    return io_.get();
+  }
+  // Can be called from any thread
+  void RequestIoStart();
+
  private:
   node::Environment* parent_env_;
   std::unique_ptr<NodeInspectorClient> inspector_;

src/inspector_io.cc

@@ -20,6 +20,7 @@
 namespace node {
 namespace inspector {
 namespace {
+using AsyncAndAgent = std::pair<uv_async_t, Agent*>;
 using v8_inspector::StringBuffer;
 using v8_inspector::StringView;
 
@@ -96,6 +97,12 @@ int CloseAsyncAndLoop(uv_async_t* async) {
   return uv_loop_close(async->loop);
 }
 
+void ReleasePairOnAsyncClose(uv_handle_t* async) {
+  AsyncAndAgent* pair = node::ContainerOf(&AsyncAndAgent::first,
+                                          reinterpret_cast<uv_async_t*>(async));
+  delete pair;
+}
+
 }  // namespace
 
 std::unique_ptr<StringBuffer> Utf8ToStringView(const std::string& message) {
@@ -127,6 +134,9 @@ class InspectorIoDelegate: public node::inspector::SocketServerDelegate {
   std::string GetTargetTitle(const std::string& id) override;
   std::string GetTargetUrl(const std::string& id) override;
   bool IsConnected() { return connected_; }
+  void ServerDone() override {
+    io_->ServerDone();
+  }
  private:
   InspectorIo* io_;
   bool connected_;
@@ -137,53 +147,70 @@ class InspectorIoDelegate: public node::inspector::SocketServerDelegate {
   bool waiting_;
 };
 
-void InterruptCallback(v8::Isolate*, void* io) {
-  static_cast<InspectorIo*>(io)->DispatchMessages();
+void InterruptCallback(v8::Isolate*, void* agent) {
+  InspectorIo* io = static_cast<Agent*>(agent)->io();
+  if (io != nullptr)
+    io->DispatchMessages();
 }
 
-class DispatchOnInspectorBackendTask : public v8::Task {
+class DispatchMessagesTask : public v8::Task {
  public:
-  explicit DispatchOnInspectorBackendTask(InspectorIo* io) : io_(io) {}
+  explicit DispatchMessagesTask(Agent* agent) : agent_(agent) {}
 
   void Run() override {
-    io_->DispatchMessages();
+    InspectorIo* io = agent_->io();
+    if (io != nullptr)
+      io->DispatchMessages();
   }
 
  private:
-  InspectorIo* io_;
+  Agent* agent_;
 };
 
 InspectorIo::InspectorIo(Environment* env, v8::Platform* platform,
-                         const std::string& path, const DebugOptions& options)
+                         const std::string& path, const DebugOptions& options,
+                         bool wait_for_connect)
                          : options_(options), thread_(), delegate_(nullptr),
-                           shutting_down_(false), state_(State::kNew),
-                           parent_env_(env), io_thread_req_(),
-                           platform_(platform), dispatching_messages_(false),
-                           session_id_(0), script_name_(path) {
-  CHECK_EQ(0, uv_async_init(env->event_loop(), &main_thread_req_,
+                           state_(State::kNew), parent_env_(env),
+                           io_thread_req_(), platform_(platform),
+                           dispatching_messages_(false), session_id_(0),
+                           script_name_(path),
+                           wait_for_connect_(wait_for_connect) {
+  main_thread_req_ = new AsyncAndAgent({uv_async_t(), env->inspector_agent()});
+  CHECK_EQ(0, uv_async_init(env->event_loop(), &main_thread_req_->first,
                             InspectorIo::MainThreadAsyncCb));
-  uv_unref(reinterpret_cast<uv_handle_t*>(&main_thread_req_));
+  uv_unref(reinterpret_cast<uv_handle_t*>(&main_thread_req_->first));
   CHECK_EQ(0, uv_sem_init(&start_sem_, 0));
 }
 
+InspectorIo::~InspectorIo() {
+  uv_sem_destroy(&start_sem_);
+  uv_close(reinterpret_cast<uv_handle_t*>(&main_thread_req_->first),
+           ReleasePairOnAsyncClose);
+}
+
 bool InspectorIo::Start() {
+  CHECK_EQ(state_, State::kNew);
   CHECK_EQ(uv_thread_create(&thread_, InspectorIo::ThreadCbIO, this), 0);
   uv_sem_wait(&start_sem_);
 
   if (state_ == State::kError) {
-    Stop();
     return false;
   }
   state_ = State::kAccepting;
-  if (options_.wait_for_connect()) {
+  if (wait_for_connect_) {
     DispatchMessages();
   }
   return true;
 }
 
 void InspectorIo::Stop() {
+  CHECK(state_ == State::kAccepting || state_ == State::kConnected);
+  Write(TransportAction::kKill, 0, StringView());
   int err = uv_thread_join(&thread_);
   CHECK_EQ(err, 0);
+  state_ = State::kShutDown;
+  DispatchMessages();
 }
 
 bool InspectorIo::IsConnected() {
@@ -195,8 +222,10 @@ bool InspectorIo::IsStarted() {
 }
 
 void InspectorIo::WaitForDisconnect() {
+  if (state_ == State::kAccepting)
+    state_ = State::kDone;
   if (state_ == State::kConnected) {
-    shutting_down_ = true;
+    state_ = State::kShutDown;
     Write(TransportAction::kStop, 0, StringView());
     fprintf(stderr, "Waiting for the debugger to disconnect...\n");
     fflush(stderr);
@@ -222,6 +251,9 @@ void InspectorIo::WriteCbIO(uv_async_t* async) {
   io->SwapBehindLock(&io->outgoing_message_queue_, &outgoing_messages);
   for (const auto& outgoing : outgoing_messages) {
     switch (std::get<0>(outgoing)) {
+    case TransportAction::kKill:
+      io_and_transport->first->TerminateConnections();
+      // Fallthrough
     case TransportAction::kStop:
       io_and_transport->first->Stop(nullptr);
       break;
@@ -253,7 +285,7 @@ void InspectorIo::WorkerRunIO() {
     uv_fs_req_cleanup(&req);
   }
   InspectorIoDelegate delegate(this, script_path, script_name_,
-                               options_.wait_for_connect());
+                               wait_for_connect_);
   delegate_ = &delegate;
   InspectorSocketServer server(&delegate,
                                options_.host_name(),
@@ -266,14 +298,12 @@ void InspectorIo::WorkerRunIO() {
     uv_sem_post(&start_sem_);
     return;
   }
-  if (!options_.wait_for_connect()) {
+  if (!wait_for_connect_) {
     uv_sem_post(&start_sem_);
   }
   uv_run(&loop, UV_RUN_DEFAULT);
   io_thread_req_.data = nullptr;
-  server.Stop(nullptr);
-  server.TerminateConnections(nullptr);
-  CHECK_EQ(CloseAsyncAndLoop(&io_thread_req_), 0);
+  CHECK_EQ(uv_loop_close(&loop), 0);
   delegate_ = nullptr;
 }
 
@@ -298,11 +328,12 @@ void InspectorIo::PostIncomingMessage(InspectorAction action, int session_id,
                                       const std::string& message) {
   if (AppendMessage(&incoming_message_queue_, action, session_id,
                     Utf8ToStringView(message))) {
+    Agent* agent = main_thread_req_->second;
     v8::Isolate* isolate = parent_env_->isolate();
     platform_->CallOnForegroundThread(isolate,
-                                      new DispatchOnInspectorBackendTask(this));
-    isolate->RequestInterrupt(InterruptCallback, this);
-    CHECK_EQ(0, uv_async_send(&main_thread_req_));
+                                      new DispatchMessagesTask(agent));
+    isolate->RequestInterrupt(InterruptCallback, agent);
+    CHECK_EQ(0, uv_async_send(&main_thread_req_->first));
   }
   NotifyMessageReceived();
 }
@@ -344,7 +375,7 @@ void InspectorIo::DispatchMessages() {
         break;
       case InspectorAction::kEndSession:
         CHECK_NE(session_delegate_, nullptr);
-        if (shutting_down_) {
+        if (state_ == State::kShutDown) {
           state_ = State::kDone;
         } else {
           state_ = State::kAccepting;
@@ -363,12 +394,18 @@ void InspectorIo::DispatchMessages() {
 
 // static
 void InspectorIo::MainThreadAsyncCb(uv_async_t* req) {
-  InspectorIo* io = node::ContainerOf(&InspectorIo::main_thread_req_, req);
-  io->DispatchMessages();
+  AsyncAndAgent* pair = node::ContainerOf(&AsyncAndAgent::first, req);
+  // Note that this may be called after io was closed or even after a new
+  // one was created and ran.
+  InspectorIo* io = pair->second->io();
+  if (io != nullptr)
+    io->DispatchMessages();
 }
 
 void InspectorIo::Write(TransportAction action, int session_id,
                         const StringView& inspector_message) {
+  if (state_ == State::kShutDown)
+    return;
   AppendMessage(&outgoing_message_queue_, action, session_id,
                 StringBuffer::create(inspector_message));
   int err = uv_async_send(&io_thread_req_);

src/inspector_io.h

@@ -31,18 +31,25 @@ namespace inspector {
 class InspectorIoDelegate;
 
 enum class InspectorAction {
-  kStartSession, kEndSession, kSendMessage
+  kStartSession,
+  kEndSession,
+  kSendMessage
 };
 
+// kKill closes connections and stops the server, kStop only stops the server
 enum class TransportAction {
-  kSendMessage, kStop
+  kKill,
+  kSendMessage,
+  kStop
 };
 
 class InspectorIo {
  public:
   InspectorIo(node::Environment* env, v8::Platform* platform,
-              const std::string& path, const DebugOptions& options);
+              const std::string& path, const DebugOptions& options,
+              bool wait_for_connect);
 
+  ~InspectorIo();
   // Start the inspector agent thread
   bool Start();
   // Stop the inspector agent
@@ -57,13 +64,23 @@ class InspectorIo {
   void ResumeStartup() {
     uv_sem_post(&start_sem_);
   }
+  void ServerDone() {
+    uv_close(reinterpret_cast<uv_handle_t*>(&io_thread_req_), nullptr);
+  }
 
  private:
   template <typename Action>
   using MessageQueue =
       std::vector<std::tuple<Action, int,
                   std::unique_ptr<v8_inspector::StringBuffer>>>;
-  enum class State { kNew, kAccepting, kConnected, kDone, kError };
+  enum class State {
+    kNew,
+    kAccepting,
+    kConnected,
+    kDone,
+    kError,
+    kShutDown
+  };
 
   static void ThreadCbIO(void* agent);
   static void MainThreadAsyncCb(uv_async_t* req);
@@ -94,12 +111,13 @@ class InspectorIo {
   uv_thread_t thread_;
 
   InspectorIoDelegate* delegate_;
-  bool shutting_down_;
   State state_;
   node::Environment* parent_env_;
 
   uv_async_t io_thread_req_;
-  uv_async_t main_thread_req_;
+  // Note that this will live while the async is being closed - likely, past
+  // the parent object lifespan
+  std::pair<uv_async_t, Agent*>* main_thread_req_;
   std::unique_ptr<InspectorSessionDelegate> session_delegate_;
   v8::Platform* platform_;
   MessageQueue<InspectorAction> incoming_message_queue_;
@@ -110,8 +128,9 @@ class InspectorIo {
   std::string script_name_;
   std::string script_path_;
   const std::string id_;
+  const bool wait_for_connect_;
 
-  friend class DispatchOnInspectorBackendTask;
+  friend class DispatchMessagesTask;
   friend class IoSessionDelegate;
   friend void InterruptCallback(v8::Isolate*, void* agent);
 };

src/inspector_socket_server.cc

@@ -212,7 +212,7 @@ class Closer {
 class SocketSession {
  public:
   SocketSession(InspectorSocketServer* server, int id);
-  void Close(bool socket_cleanup, Closer* closer);
+  void Close();
   void Declined() { state_ = State::kDeclined; }
   static SocketSession* From(InspectorSocket* socket) {
     return node::ContainerOf(&SocketSession::socket_, socket);
@@ -233,10 +233,8 @@ class SocketSession {
   static void CloseCallback_(InspectorSocket* socket, int code);
   static void ReadCallback_(uv_stream_t* stream, ssize_t read,
                             const uv_buf_t* buf);
-  void OnRemoteDataIO(InspectorSocket* socket, ssize_t read,
-                      const uv_buf_t* buf);
+  void OnRemoteDataIO(ssize_t read, const uv_buf_t* buf);
   const int id_;
-  Closer* closer_;
   InspectorSocket socket_;
   InspectorSocketServer* server_;
   std::string target_id_;
@@ -253,7 +251,9 @@ InspectorSocketServer::InspectorSocketServer(SocketServerDelegate* delegate,
                                                           server_(uv_tcp_t()),
                                                           closer_(nullptr),
                                                           next_session_id_(0),
-                                                          out_(out) { }
+                                                          out_(out) {
+  state_ = ServerState::kNew;
+}
 
 
 // static
@@ -271,7 +271,7 @@ bool InspectorSocketServer::HandshakeCallback(InspectorSocket* socket,
     SocketSession::From(socket)->FrontendConnected();
     return true;
   case kInspectorHandshakeFailed:
-    SocketSession::From(socket)->Close(false, nullptr);
+    server->SessionTerminated(SocketSession::From(socket));
     return false;
   default:
     UNREACHABLE();
@@ -294,15 +294,21 @@ bool InspectorSocketServer::SessionStarted(SocketSession* session,
   return connected;
 }
 
-void InspectorSocketServer::SessionTerminated(int session_id) {
-  if (connected_sessions_.erase(session_id) == 0) {
-    return;
-  }
-  delegate_->EndSession(session_id);
-  if (connected_sessions_.empty() &&
-      uv_is_active(reinterpret_cast<uv_handle_t*>(&server_))) {
-    PrintDebuggerReadyMessage(host_, port_, delegate_->GetTargetIds(), out_);
+void InspectorSocketServer::SessionTerminated(SocketSession* session) {
+  int id = session->Id();
+  if (connected_sessions_.erase(id) != 0) {
+    delegate_->EndSession(id);
+    if (connected_sessions_.empty()) {
+      if (state_ == ServerState::kRunning) {
+        PrintDebuggerReadyMessage(host_, port_,
+                                  delegate_->GetTargetIds(), out_);
+      }
+      if (state_ == ServerState::kStopped) {
+        delegate_->ServerDone();
+      }
+    }
   }
+  delete session;
 }
 
 bool InspectorSocketServer::RespondToGet(InspectorSocket* socket,
@@ -369,6 +375,7 @@ void InspectorSocketServer::SendListResponse(InspectorSocket* socket) {
 }
 
 bool InspectorSocketServer::Start(uv_loop_t* loop) {
+  CHECK_EQ(state_, ServerState::kNew);
   loop_ = loop;
   sockaddr_in addr;
   uv_tcp_init(loop_, &server_);
@@ -382,6 +389,7 @@ bool InspectorSocketServer::Start(uv_loop_t* loop) {
                     SocketConnectedCallback);
   }
   if (err == 0 && connected_sessions_.empty()) {
+    state_ = ServerState::kRunning;
     PrintDebuggerReadyMessage(host_, port_, delegate_->GetTargetIds(), out_);
   }
   if (err != 0 && connected_sessions_.empty()) {
@@ -397,32 +405,21 @@ bool InspectorSocketServer::Start(uv_loop_t* loop) {
 }
 
 void InspectorSocketServer::Stop(ServerCallback cb) {
+  CHECK_EQ(state_, ServerState::kRunning);
   if (closer_ == nullptr) {
     closer_ = new Closer(this);
   }
   closer_->AddCallback(cb);
-
-  uv_handle_t* handle = reinterpret_cast<uv_handle_t*>(&server_);
-  if (uv_is_active(handle)) {
-    closer_->IncreaseExpectedCount();
-    uv_close(reinterpret_cast<uv_handle_t*>(&server_), ServerClosedCallback);
-  }
+  closer_->IncreaseExpectedCount();
+  state_ = ServerState::kStopping;
+  uv_close(reinterpret_cast<uv_handle_t*>(&server_), ServerClosedCallback);
   closer_->NotifyIfDone();
 }
 
-void InspectorSocketServer::TerminateConnections(ServerCallback cb) {
-  if (closer_ == nullptr) {
-    closer_ = new Closer(this);
-  }
-  closer_->AddCallback(cb);
-  std::map<int, SocketSession*> sessions;
-  std::swap(sessions, connected_sessions_);
-  for (const auto& session : sessions) {
-    int id = session.second->Id();
-    session.second->Close(true, closer_);
-    delegate_->EndSession(id);
+void InspectorSocketServer::TerminateConnections() {
+  for (const auto& session : connected_sessions_) {
+    session.second->Close();
   }
-  closer_->NotifyIfDone();
 }
 
 bool InspectorSocketServer::TargetExists(const std::string& id) {
@@ -441,8 +438,14 @@ void InspectorSocketServer::Send(int session_id, const std::string& message) {
 // static
 void InspectorSocketServer::ServerClosedCallback(uv_handle_t* server) {
   InspectorSocketServer* socket_server = InspectorSocketServer::From(server);
-  if (socket_server->closer_)
+  CHECK_EQ(socket_server->state_, ServerState::kStopping);
+  if (socket_server->closer_) {
     socket_server->closer_->DecreaseExpectedCount();
+  }
+  if (socket_server->connected_sessions_.empty()) {
+    socket_server->delegate_->ServerDone();
+  }
+  socket_server->state_ = ServerState::kStopped;
 }
 
 // static
@@ -461,32 +464,20 @@ void InspectorSocketServer::SocketConnectedCallback(uv_stream_t* server,
 
 // InspectorSession tracking
 SocketSession::SocketSession(InspectorSocketServer* server, int id)
-                             : id_(id), closer_(nullptr), server_(server),
+                             : id_(id), server_(server),
                                state_(State::kHttp) { }
 
-void SocketSession::Close(bool socket_cleanup, Closer* closer) {
-  CHECK_EQ(closer_, nullptr);
+void SocketSession::Close() {
   CHECK_NE(state_, State::kClosing);
-  server_->SessionTerminated(id_);
-  if (socket_cleanup) {
-    state_ = State::kClosing;
-    closer_ = closer;
-    if (closer_ != nullptr)
-      closer->IncreaseExpectedCount();
-    inspector_close(&socket_, CloseCallback_);
-  } else {
-    delete this;
-  }
+  state_ = State::kClosing;
+  inspector_close(&socket_, CloseCallback_);
 }
 
 // static
 void SocketSession::CloseCallback_(InspectorSocket* socket, int code) {
   SocketSession* session = SocketSession::From(socket);
   CHECK_EQ(State::kClosing, session->state_);
-  Closer* closer = session->closer_;
-  if (closer != nullptr)
-    closer->DecreaseExpectedCount();
-  delete session;
+  session->server_->SessionTerminated(session);
 }
 
 void SocketSession::FrontendConnected() {
@@ -499,16 +490,14 @@ void SocketSession::FrontendConnected() {
 void SocketSession::ReadCallback_(uv_stream_t* stream, ssize_t read,
                                   const uv_buf_t* buf) {
   InspectorSocket* socket = inspector_from_stream(stream);
-  SocketSession::From(socket)->OnRemoteDataIO(socket, read, buf);
+  SocketSession::From(socket)->OnRemoteDataIO(read, buf);
 }
 
-void SocketSession::OnRemoteDataIO(InspectorSocket* socket, ssize_t read,
-                                   const uv_buf_t* buf) {
+void SocketSession::OnRemoteDataIO(ssize_t read, const uv_buf_t* buf) {
   if (read > 0) {
     server_->Delegate()->MessageReceived(id_, std::string(buf->base, read));
   } else {
-    server_->SessionTerminated(id_);
-    Close(true, nullptr);
+    Close();
   }
   if (buf != nullptr && buf->base != nullptr)
     delete[] buf->base;

src/inspector_socket_server.h

@@ -27,6 +27,7 @@ class SocketServerDelegate {
   virtual std::vector<std::string> GetTargetIds() = 0;
   virtual std::string GetTargetTitle(const std::string& id) = 0;
   virtual std::string GetTargetUrl(const std::string& id) = 0;
+  virtual void ServerDone() = 0;
 };
 
 class InspectorSocketServer {
@@ -39,7 +40,7 @@ class InspectorSocketServer {
   bool Start(uv_loop_t* loop);
   void Stop(ServerCallback callback);
   void Send(int session_id, const std::string& message);
-  void TerminateConnections(ServerCallback callback);
+  void TerminateConnections();
   int port() {
     return port_;
   }
@@ -59,11 +60,11 @@ class InspectorSocketServer {
   void SendListResponse(InspectorSocket* socket);
   void ReadCallback(InspectorSocket* socket, ssize_t read, const uv_buf_t* buf);
   bool SessionStarted(SocketSession* session, const std::string& id);
-  void SessionTerminated(int id);
+  void SessionTerminated(SocketSession* session);
   bool TargetExists(const std::string& id);
-  static void SocketSessionDeleter(SocketSession*);
   SocketServerDelegate* Delegate() { return delegate_; }
 
+  enum class ServerState {kNew, kRunning, kStopping, kStopped};
   uv_loop_t* loop_;
   SocketServerDelegate* const delegate_;
   const std::string host_;
@@ -74,6 +75,7 @@ class InspectorSocketServer {
   std::map<int, SocketSession*> connected_sessions_;
   int next_session_id_;
   FILE* out_;
+  ServerState state_;
 
   friend class SocketSession;
   friend class Closer;

src/node.cc

@@ -232,7 +232,6 @@ bool v8_initialized = false;
 
 // process-relative uptime base, initialized at start-up
 static double prog_start_time;
-static bool debugger_running;
 
 static Mutex node_isolate_mutex;
 static v8::Isolate* node_isolate;
@@ -261,6 +260,10 @@ static struct {
                       const node::DebugOptions& options) {
     return env->inspector_agent()->Start(platform_, script_path, options);
   }
+
+  bool InspectorStarted(Environment *env) {
+    return env->inspector_agent()->IsStarted();
+  }
 #endif  // HAVE_INSPECTOR
 
   void StartTracingAgent() {
@@ -290,6 +293,9 @@ static struct {
                     "so event tracing is not available.\n");
   }
   void StopTracingAgent() {}
+  bool InspectorStarted(Environment *env) {
+    return false;
+  }
 #endif  // !NODE_USE_V8_PLATFORM
 } v8_platform;
 
@@ -3972,9 +3978,9 @@ static void ParseArgs(int* argc,
 
 static void StartDebug(Environment* env, const char* path,
                        DebugOptions debug_options) {
-  CHECK(!debugger_running);
 #if HAVE_INSPECTOR
-  debugger_running = v8_platform.StartInspector(env, path, debug_options);
+  CHECK(!env->inspector_agent()->IsStarted());
+  v8_platform.StartInspector(env, path, debug_options);
 #endif  // HAVE_INSPECTOR
 }
 
@@ -4119,13 +4125,12 @@ static void DebugPause(const FunctionCallbackInfo<Value>& args) {
 
 
 static void DebugEnd(const FunctionCallbackInfo<Value>& args) {
-  if (debugger_running) {
 #if HAVE_INSPECTOR
-    Environment* env = Environment::GetCurrent(args);
+  Environment* env = Environment::GetCurrent(args);
+  if (env->inspector_agent()->IsStarted()) {
     env->inspector_agent()->Stop();
-#endif
-    debugger_running = false;
   }
+#endif
 }
 
 
@@ -4462,7 +4467,7 @@ inline int Start(Isolate* isolate, IsolateData* isolate_data,
   const char* path = argc > 1 ? argv[1] : nullptr;
   StartDebug(&env, path, debug_options);
 
-  if (debug_options.inspector_enabled() && !debugger_running)
+  if (debug_options.inspector_enabled() && !v8_platform.InspectorStarted(&env))
     return 12;  // Signal internal error.
 
   env.set_abort_on_uncaught_exception(abort_on_uncaught_exception);

test/cctest/test_inspector_socket_server.cc

@@ -85,7 +85,7 @@ class InspectorSocketServerTest : public ::testing::Test {
 
 class TestInspectorServerDelegate : public SocketServerDelegate {
  public:
-  TestInspectorServerDelegate() : connected(0), disconnected(0),
+  TestInspectorServerDelegate() : connected(0), disconnected(0), done(false),
                                   targets_({ MAIN_TARGET_ID,
                                              UNCONNECTABLE_TARGET_ID }),
                                   session_id_(0) {}
@@ -139,8 +139,13 @@ class TestInspectorServerDelegate : public SocketServerDelegate {
     server_->Send(session_id_, message);
   }
 
+  void ServerDone() {
+    done = true;
+  }
+
   int connected;
   int disconnected;
+  bool done;
 
  private:
   const std::vector<std::string> targets_;
@@ -302,7 +307,7 @@ class ServerHolder {
  public:
   template <typename Delegate>
   ServerHolder(Delegate* delegate, int port, FILE* out = NULL)
-               : closed(false), paused(false), sessions_terminated(false),
+               : closed(false), paused(false),
                  server_(delegate, HOST, port, out) {
     delegate->Connect(&server_);
   }
@@ -320,11 +325,6 @@ class ServerHolder {
     holder->closed = true;
   }
 
-  static void ConnectionsTerminated(InspectorSocketServer* server) {
-    ServerHolder* holder = node::ContainerOf(&ServerHolder::server_, server);
-    holder->sessions_terminated = true;
-  }
-
   static void PausedCallback(InspectorSocketServer* server) {
     ServerHolder* holder = node::ContainerOf(&ServerHolder::server_, server);
     holder->paused = true;
@@ -332,7 +332,6 @@ class ServerHolder {
 
   bool closed;
   bool paused;
-  bool sessions_terminated;
 
  private:
   InspectorSocketServer server_;
@@ -359,6 +358,12 @@ class ServerDelegateNoTargets : public SocketServerDelegate {
   std::string GetTargetUrl(const std::string& id) override {
     return "";
   }
+
+  void ServerDone() override {
+    done = true;
+  }
+
+  bool done = false;
 };
 
 static void TestHttpRequest(int port, const std::string& path,
@@ -446,7 +451,7 @@ TEST_F(InspectorSocketServerTest, InspectorSessions) {
   stays_till_termination_socket.Write(WsHandshakeRequest(MAIN_TARGET_ID));
   stays_till_termination_socket.Expect(WS_HANDSHAKE_RESPONSE);
 
-  EXPECT_EQ(3, delegate.connected);
+  SPIN_WHILE(3 != delegate.connected);
 
   delegate.Write("5678");
   stays_till_termination_socket.Expect("\x81\x4" "5678");
@@ -456,12 +461,12 @@ TEST_F(InspectorSocketServerTest, InspectorSessions) {
   delegate.Expect("1234");
 
   server->Stop(ServerHolder::CloseCallback);
-  server->TerminateConnections(ServerHolder::ConnectionsTerminated);
+  server->TerminateConnections();
 
   stays_till_termination_socket.Write(CLIENT_CLOSE_FRAME);
   stays_till_termination_socket.Expect(SERVER_CLOSE_FRAME);
 
-  EXPECT_EQ(3, delegate.disconnected);
+  SPIN_WHILE(3 != delegate.disconnected);
 
   SPIN_WHILE(!server.closed);
   stays_till_termination_socket.ExpectEOF();
@@ -473,8 +478,9 @@ TEST_F(InspectorSocketServerTest, ServerDoesNothing) {
   ASSERT_TRUE(server->Start(&loop));
 
   server->Stop(ServerHolder::CloseCallback);
-  server->TerminateConnections(ServerHolder::ConnectionsTerminated);
+  server->TerminateConnections();
   SPIN_WHILE(!server.closed);
+  ASSERT_TRUE(delegate.done);
 }
 
 TEST_F(InspectorSocketServerTest, ServerWithoutTargets) {
@@ -491,7 +497,7 @@ TEST_F(InspectorSocketServerTest, ServerWithoutTargets) {
   socket.Expect("HTTP/1.0 400 Bad Request");
   socket.ExpectEOF();
   server->Stop(ServerHolder::CloseCallback);
-  server->TerminateConnections(ServerHolder::ConnectionsTerminated);
+  server->TerminateConnections();
   SPIN_WHILE(!server.closed);
 }
 
@@ -502,11 +508,9 @@ TEST_F(InspectorSocketServerTest, ServerCannotStart) {
   ServerHolder server2(&delegate2, server1.port());
   ASSERT_FALSE(server2->Start(&loop));
   server1->Stop(ServerHolder::CloseCallback);
-  server1->TerminateConnections(ServerHolder::ConnectionsTerminated);
-  server2->Stop(ServerHolder::CloseCallback);
-  server2->TerminateConnections(ServerHolder::ConnectionsTerminated);
+  server1->TerminateConnections();
   SPIN_WHILE(!server1.closed);
-  SPIN_WHILE(!server2.closed);
+  ASSERT_TRUE(delegate1.done);
 }
 
 TEST_F(InspectorSocketServerTest, StoppingServerDoesNotKillConnections) {
@@ -521,4 +525,53 @@ TEST_F(InspectorSocketServerTest, StoppingServerDoesNotKillConnections) {
   socket1.TestHttpRequest("/json/list", "[ ]");
   socket1.Close();
   uv_run(&loop, UV_RUN_DEFAULT);
+  ASSERT_TRUE(delegate.done);
+}
+
+TEST_F(InspectorSocketServerTest, ClosingConnectionReportsDone) {
+  ServerDelegateNoTargets delegate;
+  ServerHolder server(&delegate, 0);
+  ASSERT_TRUE(server->Start(&loop));
+  SocketWrapper socket1(&loop);
+  socket1.Connect(HOST, server.port());
+  socket1.TestHttpRequest("/json/list", "[ ]");
+  server->Stop(ServerHolder::CloseCallback);
+  SPIN_WHILE(!server.closed);
+  socket1.TestHttpRequest("/json/list", "[ ]");
+  socket1.Close();
+  uv_run(&loop, UV_RUN_DEFAULT);
+  ASSERT_TRUE(delegate.done);
+}
+
+TEST_F(InspectorSocketServerTest, ClosingSocketReportsDone) {
+  TestInspectorServerDelegate delegate;
+  ServerHolder server(&delegate, 0);
+  ASSERT_TRUE(server->Start(&loop));
+  SocketWrapper socket1(&loop);
+  socket1.Connect(HOST, server.port());
+  socket1.Write(WsHandshakeRequest(MAIN_TARGET_ID));
+  socket1.Expect(WS_HANDSHAKE_RESPONSE);
+  server->Stop(ServerHolder::CloseCallback);
+  SPIN_WHILE(!server.closed);
+  ASSERT_FALSE(delegate.done);
+  socket1.Close();
+  SPIN_WHILE(!delegate.done);
+}
+
+TEST_F(InspectorSocketServerTest, TerminatingSessionReportsDone) {
+  TestInspectorServerDelegate delegate;
+  ServerHolder server(&delegate, 0);
+  ASSERT_TRUE(server->Start(&loop));
+  SocketWrapper socket1(&loop);
+  socket1.Connect(HOST, server.port());
+  socket1.Write(WsHandshakeRequest(MAIN_TARGET_ID));
+  socket1.Expect(WS_HANDSHAKE_RESPONSE);
+  server->Stop(ServerHolder::CloseCallback);
+  SPIN_WHILE(!server.closed);
+  ASSERT_FALSE(delegate.done);
+  server->TerminateConnections();
+  socket1.Expect(SERVER_CLOSE_FRAME);
+  socket1.Write(CLIENT_CLOSE_FRAME);
+  socket1.ExpectEOF();
+  SPIN_WHILE(!delegate.done);
 }

test/inspector/inspector-helper.js

@@ -8,9 +8,8 @@ const spawn = require('child_process').spawn;
 const url = require('url');
 
 const DEBUG = false;
-
 const TIMEOUT = 15 * 1000;
-
+const EXPECT_ALIVE_SYMBOL = Symbol('isAlive');
 const mainScript = path.join(common.fixturesDir, 'loop.js');
 
 function send(socket, message, id, callback) {
@@ -24,6 +23,7 @@ function send(socket, message, id, callback) {
   wsHeaderBuf.writeUInt8(0x81, 0);
   let byte2 = 0x80;
   const bodyLen = messageBuf.length;
+
   let maskOffset = 2;
   if (bodyLen < 126) {
     byte2 = 0x80 + bodyLen;
@@ -47,9 +47,17 @@ function send(socket, message, id, callback) {
       callback);
 }
 
+function sendEnd(socket) {
+  socket.write(Buffer.from([0x88, 0x80, 0x2D, 0x0E, 0x1E, 0xFA]));
+}
+
 function parseWSFrame(buffer, handler) {
   if (buffer.length < 2)
     return 0;
+  if (buffer[0] === 0x88 && buffer[1] === 0x00) {
+    handler(null);
+    return 2;
+  }
   assert.strictEqual(0x81, buffer[0]);
   let dataLen = 0x7F & buffer[1];
   let bodyOffset = 2;
@@ -136,6 +144,7 @@ function TestSession(socket, harness) {
   this.messages_ = {};
   this.expectedId_ = 1;
   this.lastMessageResponseCallback_ = null;
+  this.closeCallback_ = null;
 
   let buffer = Buffer.alloc(0);
   socket.on('data', (data) => {
@@ -146,7 +155,10 @@ function TestSession(socket, harness) {
       if (consumed)
         buffer = buffer.slice(consumed);
     } while (consumed);
-  }).on('close', () => assert(this.expectClose_, 'Socket closed prematurely'));
+  }).on('close', () => {
+    assert(this.expectClose_, 'Socket closed prematurely');
+    this.closeCallback_ && this.closeCallback_();
+  });
 }
 
 TestSession.prototype.scriptUrlForId = function(id) {
@@ -154,6 +166,11 @@ TestSession.prototype.scriptUrlForId = function(id) {
 };
 
 TestSession.prototype.processMessage_ = function(message) {
+  if (message === null) {
+    sendEnd(this.socket_);
+    return;
+  }
+
   const method = message['method'];
   if (method === 'Debugger.scriptParsed') {
     const script = message['params'];
@@ -222,6 +239,27 @@ TestSession.prototype.sendInspectorCommands = function(commands) {
   });
 };
 
+TestSession.prototype.sendCommandsAndExpectClose = function(commands) {
+  if (!(commands instanceof Array))
+    commands = [commands];
+  return this.enqueue((callback) => {
+    let timeoutId = null;
+    let done = false;
+    this.expectClose_ = true;
+    this.closeCallback_ = function() {
+      if (timeoutId)
+        clearTimeout(timeoutId);
+      done = true;
+      callback();
+    };
+    this.sendAll_(commands, () => {
+      if (!done) {
+        timeoutId = timeout('Session still open');
+      }
+    });
+  });
+};
+
 TestSession.prototype.createCallbackWithTimeout_ = function(message) {
   const promise = new Promise((resolve) => {
     this.enqueue((callback) => {
@@ -285,11 +323,23 @@ TestSession.prototype.enqueue = function(task) {
 TestSession.prototype.disconnect = function(childDone) {
   return this.enqueue((callback) => {
     this.expectClose_ = true;
-    this.harness_.childInstanceDone =
-        this.harness_.childInstanceDone || childDone;
     this.socket_.destroy();
     console.log('[test]', 'Connection terminated');
     callback();
+  }, childDone);
+};
+
+TestSession.prototype.expectClose = function() {
+  return this.enqueue((callback) => {
+    this.expectClose_ = true;
+    callback();
+  });
+};
+
+TestSession.prototype.assertClosed = function() {
+  return this.enqueue((callback) => {
+    assert.strictEqual(this.closed_, true);
+    callback();
   });
 };
 
@@ -307,8 +357,7 @@ function Harness(port, childProcess) {
   this.mainScriptPath = mainScript;
   this.stderrFilters_ = [];
   this.process_ = childProcess;
-  this.childInstanceDone = false;
-  this.returnCode_ = null;
+  this.result_ = {};
   this.running_ = true;
 
   childProcess.stdout.on('data', makeBufferingDataCallback(
@@ -323,8 +372,7 @@ function Harness(port, childProcess) {
     this.stderrFilters_ = pending;
   }));
   childProcess.on('exit', (code, signal) => {
-    assert(this.childInstanceDone, 'Child instance died prematurely');
-    this.returnCode_ = code;
+    this.result_ = {code, signal};
     this.running_ = false;
   });
 }
@@ -338,8 +386,15 @@ Harness.prototype.addStderrFilter = function(regexp, callback) {
   });
 };
 
+Harness.prototype.assertStillAlive = function() {
+  assert.strictEqual(this.running_, true,
+                     'Child died: ' + JSON.stringify(this.result_));
+};
+
 Harness.prototype.run_ = function() {
   setImmediate(() => {
+    if (!this.task_[EXPECT_ALIVE_SYMBOL])
+      this.assertStillAlive();
     this.task_(() => {
       this.task_ = this.task_.next_;
       if (this.task_)
@@ -348,7 +403,8 @@ Harness.prototype.run_ = function() {
   });
 };
 
-Harness.prototype.enqueue_ = function(task) {
+Harness.prototype.enqueue_ = function(task, expectAlive) {
+  task[EXPECT_ALIVE_SYMBOL] = !!expectAlive;
   if (!this.task_) {
     this.task_ = task;
     this.run_();
@@ -420,16 +476,16 @@ Harness.prototype.expectShutDown = function(errorCode) {
   this.enqueue_((callback) => {
     if (this.running_) {
       const timeoutId = timeout('Have not terminated');
-      this.process_.on('exit', (code) => {
+      this.process_.on('exit', (code, signal) => {
         clearTimeout(timeoutId);
-        assert.strictEqual(errorCode, code);
+        assert.strictEqual(errorCode, code, JSON.stringify({code, signal}));
         callback();
       });
     } else {
-      assert.strictEqual(errorCode, this.returnCode_);
+      assert.strictEqual(errorCode, this.result_.code);
       callback();
     }
-  });
+  }, true);
 };
 
 Harness.prototype.kill = function() {

test/inspector/test-off-no-session.js

@@ -0,0 +1,11 @@
+'use strict';
+const common = require('../common');
+common.skipIfInspectorDisabled();
+const helper = require('./inspector-helper.js');
+
+function testStop(harness) {
+  harness.expectShutDown(42);
+}
+
+helper.startNodeForInspectorTest(testStop, '--inspect',
+                                 'process._debugEnd();process.exit(42);');

test/inspector/test-off-with-session-then-on.js

@@ -0,0 +1,24 @@
+'use strict';
+const common = require('../common');
+common.skipIfInspectorDisabled();
+const helper = require('./inspector-helper.js');
+
+function testResume(session) {
+  session.sendCommandsAndExpectClose([
+    { 'method': 'Runtime.runIfWaitingForDebugger' }
+  ]);
+}
+
+function testDisconnectSession(harness) {
+  harness
+    .runFrontendSession([
+      testResume,
+    ]).expectShutDown(42);
+}
+
+const script = 'process._debugEnd();' +
+               'process._debugProcess(process.pid);' +
+               'setTimeout(() => {console.log("Done");process.exit(42)});';
+
+helper.startNodeForInspectorTest(testDisconnectSession, '--inspect-brk',
+                                 script);