fix(inspector): release the frontend's resources when DevTools disconnects

NativeScript/inspector/InspectorServer.h

@@ -4,21 +4,30 @@
 #include <dispatch/dispatch.h>
 #include <sys/types.h>
 
+#include <cstdint>
 #include <functional>
 #include <string>
 
 namespace v8_inspector {
 
 class InspectorServer {
  public:
+  using Sender = std::function<void(const std::string&)>;
+
+  // Starts listening and returns the port. Each accepted client gets a
+  // process-unique connection id: onClientConnected hands out its sender,
+  // and onClientDisconnected fires once with the same id when the
+  // connection closes, after which that sender drops every message.
+  // One client at a time: accepting a client closes the previous one. The
+  // previous client's onClientDisconnected may still be running on its own
+  // queue when the new onClientConnected starts, and a message it had
+  // already read may still reach onMessage afterwards, so callers must
+  // check the connection id.
   static in_port_t Init(
-      std::function<void(std::function<void(const std::string&)>)>
+      std::function<void(uint64_t connectionId, Sender sender)>
           onClientConnected,
-      std::function<void(const std::string&)> onMessage);
-
- private:
-  static void Send(dispatch_io_t channel, dispatch_queue_t queue,
-                   const std::string& message);
+      std::function<void(uint64_t connectionId)> onClientDisconnected,
+      std::function<void(uint64_t connectionId, const std::string&)> onMessage);
 };
 
 }  // namespace v8_inspector

NativeScript/inspector/InspectorServer.mm

@@ -3,11 +3,126 @@
 #include <netinet/in.h>
 #include <sys/socket.h>
 
+#include <atomic>
+#include <memory>
+#include <mutex>
+
 namespace v8_inspector {
 
+namespace {
+
+std::atomic<uint64_t> gLastConnectionId{0};
+
+struct Connection {
+  uint64_t id;
+  dispatch_queue_t queue;
+  dispatch_io_t channel;
+  std::atomic<bool> open{true};
+  std::function<void(uint64_t)> onDisconnected;
+  std::function<void(uint64_t, const std::string&)> onMessage;
+};
+
+void CloseConnection(const std::shared_ptr<Connection>& connection) {
+  if (!connection->open.exchange(false)) {
+    return;
+  }
+  dispatch_io_close(connection->channel, DISPATCH_IO_STOP);
+  connection->onDisconnected(connection->id);
+}
+
+// The functions below take the connection by value: a block captures a C++
+// reference as a reference, which would dangle once the caller returns.
+
+// Calls onComplete with exactly `length` bytes, or closes the connection if
+// the peer goes away (or errors) first.
+void ReadExactly(std::shared_ptr<Connection> connection, size_t length,
+                 void (^onComplete)(dispatch_data_t data)) {
+  __block dispatch_data_t received = dispatch_data_empty;
+  dispatch_io_read(connection->channel, 0, length, connection->queue,
+                   ^(bool done, dispatch_data_t data, int error) {
+                     if (data != nullptr && dispatch_data_get_size(data) > 0) {
+                       received = dispatch_data_create_concat(received, data);
+                     }
+                     if (!done) {
+                       return;
+                     }
+                     if (error != 0 || dispatch_data_get_size(received) < length) {
+                       if (error != 0 && error != ECANCELED) {
+                         NSLog(@"InspectorServer read error: %s", strerror(error));
+                       }
+                       CloseConnection(connection);
+                       return;
+                     }
+                     onComplete(received);
+                   });
+}
+
+// Messages are a big-endian uint32 byte length followed by UTF-16LE text.
+void ReadNextMessage(std::shared_ptr<Connection> connection) {
+  ReadExactly(connection, sizeof(uint32_t), ^(dispatch_data_t header) {
+    uint32_t length = 0;
+    [(NSData*)header getBytes:&length length:sizeof(length)];
+    length = ntohl(length);
+    if (length == 0) {
+      CloseConnection(connection);
+      return;
+    }
+
+    ReadExactly(connection, length, ^(dispatch_data_t body) {
+      NSString* payload = [[NSString alloc] initWithData:(NSData*)body
+                                                encoding:NSUTF16LittleEndianStringEncoding];
+      const char* utf8 = [payload UTF8String];
+      if (utf8 != nullptr) {
+        connection->onMessage(connection->id, utf8);
+      }
+      ReadNextMessage(connection);
+    });
+  });
+}
+
+void Send(std::shared_ptr<Connection> connection, const std::string& message) {
+  if (!connection->open.load()) {
+    return;
+  }
+
+  NSString* str = [NSString stringWithUTF8String:message.c_str()];
+  NSUInteger length = [str lengthOfBytesUsingEncoding:NSUTF16LittleEndianStringEncoding];
+
+  uint8_t* buffer = (uint8_t*)malloc(length + sizeof(uint32_t));
+
+  *(uint32_t*)buffer = htonl(length);
+
+  [str getBytes:&buffer[sizeof(uint32_t)]
+           maxLength:length
+          usedLength:NULL
+            encoding:NSUTF16LittleEndianStringEncoding
+             options:0
+               range:NSMakeRange(0, str.length)
+      remainingRange:NULL];
+
+  dispatch_data_t data =
+      dispatch_data_create(buffer, length + sizeof(uint32_t), connection->queue, ^{
+        free(buffer);
+      });
+
+  dispatch_io_write(connection->channel, 0, data, connection->queue,
+                    ^(bool done, dispatch_data_t data, int error) {
+                      // ECANCELED: the write was still queued when the
+                      // connection closed.
+                      if (error == 0 || error == ECANCELED) {
+                        return;
+                      }
+                      NSLog(@"InspectorServer::Send error: %s", strerror(error));
+                      CloseConnection(connection);
+                    });
+}
+
+}  // namespace
+
 in_port_t InspectorServer::Init(
-    std::function<void(std::function<void(const std::string&)>)> onClientConnected,
-    std::function<void(const std::string&)> onMessage) {
+    std::function<void(uint64_t connectionId, Sender sender)> onClientConnected,
+    std::function<void(uint64_t connectionId)> onClientDisconnected,
+    std::function<void(uint64_t connectionId, const std::string&)> onMessage) {
   in_port_t listenPort = 18183;
 
   int serverSocket = -1;
@@ -31,6 +146,12 @@
     assert(false);
   }
 
+  struct ActiveConnection {
+    std::mutex mutex;
+    std::weak_ptr<Connection> connection;
+  };
+  auto active = std::make_shared<ActiveConnection>();
+
   // Set up the dispatch source that will alert us to new incoming connections
   dispatch_queue_t q = dispatch_queue_create("server_queue", DISPATCH_QUEUE_CONCURRENT);
   dispatch_source_t acceptSource =
@@ -43,68 +164,51 @@
       unsigned int clntLen = sizeof(echoClntAddr);
 
       // Wait for a client to connect
-      if ((clientSocket = accept(serverSocket, (struct sockaddr*)&echoClntAddr, &clntLen)) >= 0) {
-        dispatch_io_t channel =
-            dispatch_io_create(DISPATCH_IO_STREAM, clientSocket, q, ^(int error) {
-              if (error) {
-                NSLog(@"InspectorServer channel cleanup error: %s", strerror(error));
-              }
-              close(clientSocket);
-            });
-
-        onClientConnected([channel, q](std::string message) { Send(channel, q, message); });
-
-        __block dispatch_io_handler_t receiver = ^(bool done, dispatch_data_t data, int error) {
-          if (error) {
-            NSLog(@"InspectorServer read message header error: %s", strerror(error));
-          }
-
-          const void* bytes = [(NSData*)data bytes];
-          if (!bytes) {
-            dispatch_io_close(channel, DISPATCH_IO_STOP);
-            return;
-          }
-
-          uint32_t length = ntohl(*(uint32_t*)bytes);
-
-          // Configure the channel...
-          dispatch_io_set_low_water(channel, length);
+      if ((clientSocket = accept(serverSocket, (struct sockaddr*)&echoClntAddr, &clntLen)) < 0) {
+        NSLog(@"InspectorServer accept() failed");
+        continue;
+      }
 
-          // Setup read handler
-          dispatch_io_read(channel, 0, length, q, ^(bool done, dispatch_data_t data, int error) {
-            BOOL close = NO;
+      // A write racing the peer's close must fail with EPIPE, not kill the app.
+      int noSigPipe = 1;
+      setsockopt(clientSocket, SOL_SOCKET, SO_NOSIGPIPE, &noSigPipe, sizeof(noSigPipe));
+
+      auto connection = std::make_shared<Connection>();
+      connection->id = ++gLastConnectionId;
+      // Serial, so the read handlers that assemble one message never overlap.
+      connection->queue =
+          dispatch_queue_create("NativeScript.v8.inspector.connection", DISPATCH_QUEUE_SERIAL);
+      connection->onDisconnected = onClientDisconnected;
+      connection->onMessage = onMessage;
+      connection->channel =
+          dispatch_io_create(DISPATCH_IO_STREAM, clientSocket, connection->queue, ^(int error) {
             if (error) {
-              NSLog(@"InspectorServer read message body error: %s", strerror(error));
-              close = YES;
+              NSLog(@"InspectorServer channel cleanup error: %s", strerror(error));
             }
+            close(clientSocket);
+          });
 
-            const size_t size = data ? dispatch_data_get_size(data) : 0;
-            if (size) {
-              NSString* payload = [[NSString alloc] initWithData:(NSData*)data
-                                                        encoding:NSUTF16LittleEndianStringEncoding];
-
-              onMessage([payload UTF8String]);
-
-#pragma clang diagnostic push
-#pragma clang diagnostic ignored "-Warc-retain-cycles"
-              dispatch_io_read(channel, 0, 4, q, receiver);
-#pragma clang diagnostic pop
-            } else {
-              close = YES;
-            }
+      if (connection->channel == nullptr) {
+        close(clientSocket);
+        continue;
+      }
 
-            if (close) {
-              dispatch_io_close(channel, DISPATCH_IO_STOP);
-            }
+      dispatch_async(connection->queue, ^{
+        {
+          // Held across both callbacks so two clients accepted back to back
+          // cannot hand out their senders in the opposite order.
+          std::lock_guard<std::mutex> lock(active->mutex);
+          std::shared_ptr<Connection> previous = active->connection.lock();
+          active->connection = connection;
+          if (previous != nullptr) {
+            CloseConnection(previous);
+          }
+          onClientConnected(connection->id, [connection](const std::string& message) {
+            Send(connection, message);
           });
-        };
-
-        if (channel) {
-          dispatch_io_read(channel, 0, 4, q, receiver);
         }
-      } else {
-        NSLog(@"InspectorServer accept() failed");
-      }
+        ReadNextMessage(connection);
+      });
     }
   });
 
@@ -121,32 +225,4 @@
   return listenPort;
 }
 
-void InspectorServer::Send(dispatch_io_t channel, dispatch_queue_t queue,
-                           const std::string& message) {
-  NSString* str = [NSString stringWithUTF8String:message.c_str()];
-  NSUInteger length = [str lengthOfBytesUsingEncoding:NSUTF16LittleEndianStringEncoding];
-
-  uint8_t* buffer = (uint8_t*)malloc(length + sizeof(uint32_t));
-
-  *(uint32_t*)buffer = htonl(length);
-
-  [str getBytes:&buffer[sizeof(uint32_t)]
-           maxLength:length
-          usedLength:NULL
-            encoding:NSUTF16LittleEndianStringEncoding
-             options:0
-               range:NSMakeRange(0, str.length)
-      remainingRange:NULL];
-
-  dispatch_data_t data = dispatch_data_create(buffer, length + sizeof(uint32_t), queue, ^{
-    free(buffer);
-  });
-
-  dispatch_io_write(channel, 0, data, queue, ^(bool done, dispatch_data_t data, int error) {
-    if (error) {
-      NSLog(@"InspectorServer::Send error: %s", strerror(error));
-    }
-  });
-}
-
 }  // namespace v8_inspector

NativeScript/inspector/JsV8InspectorClient.h

@@ -2,7 +2,9 @@
 #define JsV8InspectorClient_h
 
 #include <dispatch/dispatch.h>
+#include <netinet/in.h>
 
+#include <atomic>
 #include <functional>
 #include <map>
 #include <mutex>
@@ -23,7 +25,6 @@ class JsV8InspectorClient : V8InspectorClient, V8Inspector::Channel {
   JsV8InspectorClient(tns::Runtime* runtime);
   void init();
   void connect(int argc, char** argv);
-  void disconnect();
   void dispatchMessage(const std::string& message);
 
   // The single instance debugging the main isolate (created when IsDebug);
@@ -64,7 +65,17 @@ class JsV8InspectorClient : V8InspectorClient, V8Inspector::Channel {
  private:
   static constexpr int contextGroupId = 1;
 
-  bool isConnected_;
+  // Queued behind the departing frontend's messages; replaces the main
+  // session when it reaches the front of the queue.
+  static constexpr const char* kResetSessionMessage =
+      "{\"__nsInternal\":\"resetSession\"}";
+
+  std::atomic<bool> isConnected_{false};
+  // Set when a reset arrives inside the pause loop (which may run on any
+  // thread holding the Locker), where the session may still be on the stack;
+  // the next drain after the loop exits performs it.
+  std::atomic<bool> pendingSessionReset_{false};
+  in_port_t listenPort_ = 0;  // main queue only
   std::unique_ptr<V8Inspector> inspector_;
   v8::Persistent<v8::Context> context_;
   std::unique_ptr<V8InspectorSession> session_;
@@ -77,6 +88,8 @@ class JsV8InspectorClient : V8InspectorClient, V8Inspector::Channel {
   dispatch_queue_t messageLoopQueue_;
   dispatch_semaphore_t messageArrived_;
   std::function<void(const std::string&)> sender_;
+  uint64_t connectionId_ =
+      0;  // id of sender_'s connection, guarded by senderMutex_
   bool isWaitingForDebugger_;
   bool hasScheduledDebugBreak_;
 
@@ -110,6 +123,11 @@ class JsV8InspectorClient : V8InspectorClient, V8Inspector::Channel {
   static JsV8InspectorClient* instance_;
 
   std::mutex senderMutex_;
+  // Serializes onFrontendConnected/onFrontendDisconnected/
+  // onFrontendMessageReceived, which run on different connection queues.
+  // Taken before senderMutex_, workerTargetsMutex_ and resourceStreamsMutex_,
+  // never while holding any of them.
+  std::mutex frontendLifecycleMutex_;
 
   // Routes a frontend message carrying a sessionId to its worker session
   // (socket thread). msgId is -1 when the message has no id.
@@ -126,8 +144,20 @@ class JsV8InspectorClient : V8InspectorClient, V8Inspector::Channel {
   void createInspectorSession();
   void notify(std::unique_ptr<StringBuffer> message);
   void notify(const std::string& message);
-  void onFrontendConnected(std::function<void(const std::string&)> sender);
-  void onFrontendMessageReceived(const std::string& message);
+  void onFrontendConnected(uint64_t connectionId,
+                           std::function<void(const std::string&)> sender);
+  void onFrontendDisconnected(uint64_t connectionId);
+  void onFrontendMessageReceived(uint64_t connectionId,
+                                 const std::string& message);
+  // Socket thread: drops the state owned by the departing frontend that does
+  // not need the main isolate (IO streams, tracing, worker sessions).
+  void ResetFrontendState();
+  // Pause loop only: resumes and defers the session swap to the next drain.
+  void DeferSessionResetFromPauseLoop();
+  void ResetMainSession();
+  void QueueSessionReset();
+  void EnqueueMessage(const std::string& message);
+  void ScheduleMessageDrain();
   std::string PumpMessage();
   static void registerDomainDispatcherCallback(
       const v8::FunctionCallbackInfo<v8::Value>& args);