Skip to content

inspector: address race conditions #8672

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
eugeneo wants to merge 1 commit into from
Closed

inspector: address race conditions #8672

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
35 changes: 21 additions & 14 deletions src/inspector_agent.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -237,12 +237,13 @@ class AgentImpl {
const String16& message);
void SwapBehindLock(MessageQueue* vector1, MessageQueue* vector2);
void PostIncomingMessage(const String16& message);
void WaitForFrontendMessage();
void NotifyMessageReceived();
State ToState(State state);

uv_sem_t start_sem_;
ConditionVariable pause_cond_;
Mutex pause_lock_;
Mutex queue_lock_;
ConditionVariable incoming_message_cond_;
Mutex state_lock_;
uv_thread_t thread_;
uv_loop_t child_loop_;

Expand Down Expand Up @@ -342,15 +343,11 @@ class V8NodeInspector : public v8_inspector::V8InspectorClient {
return;
terminated_ = false;
running_nested_loop_ = true;
agent_->DispatchMessages();
do {
{
Mutex::ScopedLock scoped_lock(agent_->pause_lock_);
agent_->pause_cond_.Wait(scoped_lock);
}
while (!terminated_) {
agent_->WaitForFrontendMessage();
while (v8::platform::PumpMessageLoop(platform_, env_->isolate()))
{}
} while (!terminated_);
}
terminated_ = false;
running_nested_loop_ = false;
}
Expand Down Expand Up @@ -633,7 +630,6 @@ bool AgentImpl::OnInspectorHandshakeIO(InspectorSocket* socket,
void AgentImpl::OnRemoteDataIO(InspectorSocket* socket,
ssize_t read,
const uv_buf_t* buf) {
Mutex::ScopedLock scoped_lock(pause_lock_);
if (read > 0) {
String16 str = String16::fromUTF8(buf->base, read);
// TODO(pfeldman): Instead of blocking execution while debugger
Expand All @@ -658,7 +654,6 @@ void AgentImpl::OnRemoteDataIO(InspectorSocket* socket,
if (buf) {
delete[] buf->base;
}
pause_cond_.Broadcast(scoped_lock);
}

// static
Expand Down Expand Up @@ -724,14 +719,14 @@ void AgentImpl::WorkerRunIO() {

bool AgentImpl::AppendMessage(MessageQueue* queue, int session_id,
const String16& message) {
Mutex::ScopedLock scoped_lock(queue_lock_);
Mutex::ScopedLock scoped_lock(state_lock_);
bool trigger_pumping = queue->empty();
queue->push_back(std::make_pair(session_id, message));
return trigger_pumping;
}

void AgentImpl::SwapBehindLock(MessageQueue* vector1, MessageQueue* vector2) {
Mutex::ScopedLock scoped_lock(queue_lock_);
Mutex::ScopedLock scoped_lock(state_lock_);
vector1->swap(*vector2);
}

Expand All @@ -743,6 +738,18 @@ void AgentImpl::PostIncomingMessage(const String16& message) {
isolate->RequestInterrupt(InterruptCallback, this);
uv_async_send(data_written_);
}
NotifyMessageReceived();
}

void AgentImpl::WaitForFrontendMessage() {
Mutex::ScopedLock scoped_lock(state_lock_);
if (incoming_message_queue_.empty())
incoming_message_cond_.Wait(scoped_lock);
}

void AgentImpl::NotifyMessageReceived() {
Mutex::ScopedLock scoped_lock(state_lock_);
incoming_message_cond_.Broadcast(scoped_lock);
}

void AgentImpl::OnInspectorConnectionIO(InspectorSocket* socket) {
Expand Down
16 changes: 12 additions & 4 deletions test/inspector/inspector-helper.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,13 +6,17 @@ const http = require('http');
const path = require('path');
const spawn = require('child_process').spawn;

const DEBUG = false;

const TIMEOUT = 15 * 1000;

const mainScript = path.join(common.fixturesDir, 'loop.js');

function send(socket, message, id, callback) {
const msg = JSON.parse(JSON.stringify(message)); // Clone!
msg['id'] = id;
if (DEBUG)
console.log('[sent]', JSON.stringify(msg));
const messageBuf = Buffer.from(JSON.stringify(msg));

const wsHeaderBuf = Buffer.allocUnsafe(16);
Expand Down Expand Up @@ -61,6 +65,8 @@ function parseWSFrame(buffer, handler) {
return 0;
const message = JSON.parse(
buffer.slice(bodyOffset, bodyOffset + dataLen).toString('utf8'));
if (DEBUG)
console.log('[received]', JSON.stringify(message));
handler(message);
return bodyOffset + dataLen;
}
Expand Down Expand Up @@ -117,7 +123,7 @@ const TestSession = function(socket, harness) {
this.expectedId_ = 1;
this.lastMessageResponseCallback_ = null;

let buffer = Buffer.from('');
let buffer = new Buffer(0);
socket.on('data', (data) => {
buffer = Buffer.concat([buffer, data]);
let consumed;
Expand Down Expand Up @@ -195,9 +201,10 @@ TestSession.prototype.sendInspectorCommands = function(commands) {
timeoutId = setTimeout(() => {
let s = '';
for (const id in this.messages_) {
s += this.messages_[id] + '\n';
s += id + ', ';
}
common.fail(s.substring(0, s.length - 1));
common.fail('Messages without response: ' +
s.substring(0, s.length - 2));
}, TIMEOUT);
});
});
Expand Down Expand Up @@ -269,6 +276,7 @@ TestSession.prototype.disconnect = function(childDone) {
this.harness_.childInstanceDone =
this.harness_.childInstanceDone || childDone;
this.socket_.end();
console.log('[test]', 'Connection terminated');
callback();
});
};
Expand All @@ -293,7 +301,7 @@ const Harness = function(port, childProcess) {
if (!filter(message)) pending.push(filter);
this.stderrFilters_ = pending;
}));
childProcess.on('close', (code, signal) => {
childProcess.on('exit', (code, signal) => {
assert(this.childInstanceDone, 'Child instance died prematurely');
this.returnCode_ = code;
this.running_ = false;
Expand Down