Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 4 additions & 6 deletions NativeScript/inspector/InspectorServer.h
Original file line number Diff line number Diff line change
Expand Up @@ -18,12 +18,10 @@ class InspectorServer {
// 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, and
// that close reports onClientDisconnected before the new onClientConnected.
// A client that closed itself (EOF, read or write error) reports from its
// own queue, which can interleave with a new onClientConnected. A message
// the previous client had already read may reach onMessage after the
// takeover, so callers must check the connection id.
// Any number of clients may be connected at once. For one connection the
// callbacks are ordered: onClientConnected returns before its first
// onMessage, and onClientDisconnected comes after its last. Callbacks for
// different connections run concurrently.
static in_port_t Init(
std::function<void(uint64_t connectionId, Sender sender)>
onClientConnected,
Expand Down
62 changes: 42 additions & 20 deletions NativeScript/inspector/InspectorServer.mm
Original file line number Diff line number Diff line change
Expand Up @@ -12,15 +12,29 @@

std::atomic<uint64_t> gLastConnectionId{0};

// A client this far behind is assumed gone (e.g. a half-open socket whose
// peer vanished without a FIN): every client receives broadcast events, so
// its unsent output would otherwise grow without bound. Large enough for a
// heap snapshot streamed over a slow link.
constexpr size_t kMaxPendingWriteBytes = 256 * 1024 * 1024;

struct Connection {
uint64_t id;
dispatch_queue_t queue;
dispatch_io_t channel;
// Only changes on `queue`.
std::atomic<bool> open{true};
std::atomic<size_t> pendingWriteBytes{0};
std::function<void(uint64_t)> onDisconnected;
std::function<void(uint64_t, const std::string&)> onMessage;
};

// The peer went away or we closed the channel ourselves (ECANCELED: an
// operation was still queued at close time); routine, not worth a log line.
bool IsDisconnectError(int error) {
return error == ECANCELED || error == ECONNRESET || error == EPIPE;
}

void CloseConnection(const std::shared_ptr<Connection>& connection) {
if (!connection->open.exchange(false)) {
return;
Expand All @@ -46,7 +60,7 @@ void ReadExactly(std::shared_ptr<Connection> connection, size_t length,
return;
}
if (error != 0 || dispatch_data_get_size(received) < length) {
if (error != 0 && error != ECANCELED) {
if (error != 0 && !IsDisconnectError(error)) {
NSLog(@"InspectorServer read error: %s", strerror(error));
}
CloseConnection(connection);
Expand All @@ -70,6 +84,11 @@ void ReadNextMessage(std::shared_ptr<Connection> connection) {
ReadExactly(connection, length, ^(dispatch_data_t body) {
NSString* payload = [[NSString alloc] initWithData:(NSData*)body
encoding:NSUTF16LittleEndianStringEncoding];
// A write error may have closed the connection after this read
// completed; onMessage must never follow onDisconnected.
if (!connection->open.load()) {
return;
}
const char* utf8 = [payload UTF8String];
if (utf8 != nullptr) {
connection->onMessage(connection->id, utf8);
Expand Down Expand Up @@ -99,19 +118,31 @@ void Send(std::shared_ptr<Connection> connection, const std::string& message) {
range:NSMakeRange(0, str.length)
remainingRange:NULL];

dispatch_data_t data =
dispatch_data_create(buffer, length + sizeof(uint32_t), connection->queue, ^{
free(buffer);
});
size_t size = length + sizeof(uint32_t);
dispatch_data_t data = dispatch_data_create(buffer, size, connection->queue, ^{
free(buffer);
});

if (connection->pendingWriteBytes.fetch_add(size) + size > kMaxPendingWriteBytes) {
NSLog(@"InspectorServer: closing a client that stopped reading");
// Closed on its own queue to keep onDisconnected ordered after onMessage.
dispatch_async(connection->queue, ^{
CloseConnection(connection);
});
return;
}

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) {
if (done) {
connection->pendingWriteBytes.fetch_sub(size);
}
if (error == 0) {
return;
}
NSLog(@"InspectorServer::Send error: %s", strerror(error));
if (!IsDisconnectError(error)) {
NSLog(@"InspectorServer::Send error: %s", strerror(error));
}
CloseConnection(connection);
});
}
Expand Down Expand Up @@ -145,10 +176,6 @@ void Send(std::shared_ptr<Connection> connection, const std::string& message) {
assert(false);
}

// Only touched by the accept handler, which libdispatch never runs
// concurrently with itself.
__block std::weak_ptr<Connection> 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 =
Expand Down Expand Up @@ -190,13 +217,8 @@ void Send(std::shared_ptr<Connection> connection, const std::string& message) {
continue;
}

// The accept handler is serial, so back-to-back accepts take over in
// order and the newest client stays active.
std::shared_ptr<Connection> previous = activeConnection.lock();
activeConnection = connection;
if (previous != nullptr) {
CloseConnection(previous);
}
// Reads start only after this returns, so the client's first message
// cannot overtake its own connect.
onClientConnected(connection->id,
[connection](const std::string& message) { Send(connection, message); });

Expand Down
82 changes: 82 additions & 0 deletions NativeScript/inspector/InspectorSessions.h
Original file line number Diff line number Diff line change
@@ -0,0 +1,82 @@
#ifndef InspectorSessions_h
#define InspectorSessions_h

#include <cstdint>
#include <functional>
#include <map>
#include <memory>
#include <mutex>
#include <string>
#include <vector>

#include "include/v8-inspector.h"

namespace v8_inspector {

// One V8InspectorSession per connected frontend, all on one isolate and
// context group. Each session's output goes only to its own frontend.
//
// Except for RequestPause and ClearPauseRequests, call only from the one
// thread at a time that feeds the isolate's inspector messages (its message
// drain or pause loop), or from an interrupt on the thread running its JS.
// The methods take the isolate's Locker themselves (v8::Locker is reentrant).
class InspectorSessions {
public:
using Output =
std::function<void(uint64_t connectionId, const std::string& message)>;

InspectorSessions(v8::Isolate* isolate, int contextGroupId, Output output);
~InspectorSessions();

// Creates the frontend's session unless it already has one.
void Connect(V8Inspector* inspector, uint64_t connectionId);

// Ignored for a frontend without a session or one being closed.
void Dispatch(uint64_t connectionId, const std::string& message);

// Destroys the frontend's session. Its destructor disables its agents, so
// a pause ends only if no other frontend still has the debugger enabled.
// A session cannot be destroyed while it is dispatching (a pause loop can
// sit under its dispatchProtocolMessage); it then gets Debugger.disable
// right away and is destroyed once that dispatch returns.
void Disconnect(uint64_t connectionId);

// Destroys every session. Never while any of them is dispatching.
void DisconnectAll();

bool Has(uint64_t connectionId) const;

// Any thread. Records that the frontend asked to pause; TakePauseRequests,
// called from an isolate interrupt, schedules the pause in its session.
void RequestPause(uint64_t connectionId);
void TakePauseRequests();
// Any thread. Called when a pause starts, so a request whose interrupt is
// still pending cannot pause again after the resume.
void ClearPauseRequests();

void SchedulePauseInAll();

private:
class Channel;
struct Session {
std::unique_ptr<Channel> channel;
std::unique_ptr<V8InspectorSession> session;
int dispatchDepth = 0;
bool closing = false;
};

void DispatchTo(Session* session, const std::string& message);
void Destroy(uint64_t connectionId);

v8::Isolate* isolate_;
int contextGroupId_;
Output output_;
std::map<uint64_t, std::unique_ptr<Session>> sessions_;

std::mutex pauseRequestsMutex_;
std::vector<uint64_t> pauseRequests_; // guarded by pauseRequestsMutex_
};

} // namespace v8_inspector

#endif /* InspectorSessions_h */
160 changes: 160 additions & 0 deletions NativeScript/inspector/InspectorSessions.mm
Original file line number Diff line number Diff line change
@@ -0,0 +1,160 @@
#include "InspectorSessions.h"

#include "include/v8-locker.h"
#include "utils.h"

using namespace v8;

namespace v8_inspector {

namespace {
StringView Make8BitStringView(const std::string& value) {
return StringView(reinterpret_cast<const uint8_t*>(value.data()), value.size());
}

constexpr const char* kDisableDebugger = "{\"id\":0,\"method\":\"Debugger.disable\"}";
} // namespace

class InspectorSessions::Channel final : public V8Inspector::Channel {
public:
Channel(const Output& output, uint64_t connectionId)
: output_(output), connectionId_(connectionId) {}

void sendResponse(int callId, std::unique_ptr<StringBuffer> message) override {
output_(connectionId_, ToStdString(message->string()));
}

void sendNotification(std::unique_ptr<StringBuffer> message) override {
output_(connectionId_, ToStdString(message->string()));
}

void flushProtocolNotifications() override {}

private:
const Output& output_;
uint64_t connectionId_;
};

InspectorSessions::InspectorSessions(Isolate* isolate, int contextGroupId, Output output)
: isolate_(isolate), contextGroupId_(contextGroupId), output_(std::move(output)) {}

InspectorSessions::~InspectorSessions() { this->DisconnectAll(); }

void InspectorSessions::Connect(V8Inspector* inspector, uint64_t connectionId) {
if (sessions_.count(connectionId) != 0) {
return;
}

Locker locker(isolate_);
Isolate::Scope isolate_scope(isolate_);
HandleScope handle_scope(isolate_);

auto entry = std::make_unique<Session>();
entry->channel = std::make_unique<Channel>(output_, connectionId);
entry->session =
inspector->connect(contextGroupId_, entry->channel.get(), {}, V8Inspector::kFullyTrusted);
sessions_.emplace(connectionId, std::move(entry));
}

void InspectorSessions::Dispatch(uint64_t connectionId, const std::string& message) {
auto it = sessions_.find(connectionId);
if (it == sessions_.end() || it->second->closing) {
return;
}

Session* session = it->second.get();
this->DispatchTo(session, message);
if (session->closing && session->dispatchDepth == 0) {
this->Destroy(connectionId);
}
}

void InspectorSessions::DispatchTo(Session* session, const std::string& message) {
Locker locker(isolate_);
Isolate::Scope isolate_scope(isolate_);
HandleScope handle_scope(isolate_);

// Destroy never runs while this is above zero, so `session` stays valid
// across the call even if the frontend disconnects inside it.
session->dispatchDepth++;
session->session->dispatchProtocolMessage(Make8BitStringView(message));
session->dispatchDepth--;
}

void InspectorSessions::Disconnect(uint64_t connectionId) {
auto it = sessions_.find(connectionId);
if (it == sessions_.end()) {
return;
}

Session* session = it->second.get();
if (session->dispatchDepth == 0) {
this->Destroy(connectionId);
return;
}

if (!session->closing) {
session->closing = true;
this->DispatchTo(session, kDisableDebugger);
}
}

void InspectorSessions::DisconnectAll() {
while (!sessions_.empty()) {
this->Destroy(sessions_.begin()->first);
}
}

bool InspectorSessions::Has(uint64_t connectionId) const {
auto it = sessions_.find(connectionId);
return it != sessions_.end() && !it->second->closing;
}

void InspectorSessions::Destroy(uint64_t connectionId) {
auto node = sessions_.extract(connectionId);
if (node.empty()) {
return;
}

Locker locker(isolate_);
Isolate::Scope isolate_scope(isolate_);
HandleScope handle_scope(isolate_);
// The session uses its channel until its destructor returns.
node.mapped()->session.reset();
node.mapped()->channel.reset();
}

void InspectorSessions::RequestPause(uint64_t connectionId) {
std::lock_guard<std::mutex> lock(pauseRequestsMutex_);
pauseRequests_.push_back(connectionId);
}

void InspectorSessions::TakePauseRequests() {
std::vector<uint64_t> requests;
{
std::lock_guard<std::mutex> lock(pauseRequestsMutex_);
requests.swap(pauseRequests_);
}

for (uint64_t connectionId : requests) {
auto it = sessions_.find(connectionId);
if (it != sessions_.end() && !it->second->closing) {
it->second->session->schedulePauseOnNextStatement({}, {});
}
}
}

void InspectorSessions::ClearPauseRequests() {
std::lock_guard<std::mutex> lock(pauseRequestsMutex_);
pauseRequests_.clear();
}

void InspectorSessions::SchedulePauseInAll() {
for (auto& entry : sessions_) {
if (!entry.second->closing) {
entry.second->session->schedulePauseOnNextStatement({}, {});
}
}
}

} // namespace v8_inspector
Loading
Loading