foster/denoland-deno
1
0
Fork 0
mirror of https://github.com/denoland/deno.git synced 2025-01-09 07:39:15 -05:00
Code Issues Packages 1 Wiki Activity

Fix inspector hanging when task budget is exceeded (#5083)

Browse source 
The issue is solved by proxying websocket messages over a pair of
`futures::mpsc::unbounded` channels. As these are are implemented in
the 'futures' crate, they can't participate in Tokio's cooperative
task yielding.
This commit is contained in:
Bert Belder 2020-05-05 05:39:42 +02:00
parent 6e287d9518
commit e574437922
No known key found for this signature in database
GPG key ID: 7A77887B2E2ED461
3 changed files with 204 additions and 71 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

142
cli/inspector.rs
View file

@ -37,7 +37,6 @@ use std::sync::Once;
use std::thread;
use uuid::Uuid;
use warp::filters::ws;
use warp::filters::ws::WebSocket;
use warp::Filter;
struct InspectorServer {
@ -91,7 +90,7 @@ struct InspectorInfo {
host: SocketAddr,
uuid: Uuid,
thread_name: Option<String>,
new_websocket_tx: UnboundedSender<WebSocket>,
new_websocket_tx: UnboundedSender<WebSocketProxy>,
canary_rx: oneshot::Receiver<Never>,
}
@ -178,7 +177,9 @@ async fn server(
g.get(&uuid).map(|info| info.new_websocket_tx.clone()).map(
|new_websocket_tx| {
ws.on_upgrade(move |websocket| async move {
let _ = new_websocket_tx.unbounded_send(websocket);
let (proxy, pump) = create_websocket_proxy(websocket);
let _ = new_websocket_tx.unbounded_send(proxy);
pump.await;
})
},
)
@ -223,6 +224,69 @@ async fn server(
}
}
type WebSocketProxySender = UnboundedSender<ws::Message>;
type WebSocketProxyReceiver =
UnboundedReceiver<Result<ws::Message, warp::Error>>;
/// Encapsulates an UnboundedSender/UnboundedReceiver pair that together form
/// a duplex channel for sending/receiving websocket messages.
struct WebSocketProxy {
tx: WebSocketProxySender,
rx: WebSocketProxyReceiver,
}
impl WebSocketProxy {
pub fn split(self) -> (WebSocketProxySender, WebSocketProxyReceiver) {
(self.tx, self.rx)
}
}
/// Creates a future that proxies messages sent and received on a warp WebSocket
/// to a UnboundedSender/UnboundedReceiver pair. We need this to sidestep
/// Tokio's task budget, which causes issues when DenoInspector::poll_sessions()
/// needs to block the thread because JavaScript execution is paused.
///
/// This works because UnboundedSender/UnboundedReceiver are implemented in the
/// 'futures' crate, therefore they can't participate in Tokio's cooperative
/// task yielding.
///
/// A tuple is returned, where the first element is a duplex channel that can
/// be used to send/receive messages on the websocket, and the second element
/// is a future that does the forwarding.
fn create_websocket_proxy(
websocket: ws::WebSocket,
) -> (WebSocketProxy, impl Future<Output = ()> + Send) {
// The 'outbound' channel carries messages sent to the websocket.
let (outbound_tx, outbound_rx) = mpsc::unbounded();
// The 'inbound' channel carries messages received from the websocket.
let (inbound_tx, inbound_rx) = mpsc::unbounded();
let proxy = WebSocketProxy {
tx: outbound_tx,
rx: inbound_rx,
};
// The pump future takes care of forwarding messages between the websocket
// and channels. It resolves to () when either side disconnects, ignoring any
// errors.
let pump = async move {
let (websocket_tx, websocket_rx) = websocket.split();
let outbound_pump =
outbound_rx.map(Ok).forward(websocket_tx).map_err(|_| ());
let inbound_pump = websocket_rx
.map(|msg| inbound_tx.unbounded_send(msg))
.map_err(|_| ())
.try_collect::<()>();
let _ = future::try_join(outbound_pump, inbound_pump).await;
};
(proxy, pump)
}
#[derive(Clone, Copy)]
enum PollState {
Idle,
@ -322,7 +386,8 @@ impl DenoInspector {
let mut hs = v8::HandleScope::new(v8_isolate);
let scope = hs.enter();
let (new_websocket_tx, new_websocket_rx) = mpsc::unbounded::<WebSocket>();
let (new_websocket_tx, new_websocket_rx) =
mpsc::unbounded::<WebSocketProxy>();
let (canary_tx, canary_rx) = oneshot::channel::<Never>();
let info = InspectorInfo {
@ -511,7 +576,7 @@ struct InspectorSessions {
impl InspectorSessions {
fn new(
inspector_ptr: *mut DenoInspector,
new_websocket_rx: UnboundedReceiver<WebSocket>,
new_websocket_rx: UnboundedReceiver<WebSocketProxy>,
) -> RefCell<Self> {
let new_incoming = new_websocket_rx
.map(move |websocket| DenoInspectorSession::new(inspector_ptr, websocket))
@ -609,11 +674,8 @@ impl task::ArcWake for InspectorWaker {
struct DenoInspectorSession {
v8_channel: v8::inspector::ChannelBase,
v8_session: v8::UniqueRef<v8::inspector::V8InspectorSession>,
message_handler: Pin<Box<dyn Future<Output = ()> + 'static>>,
// Internal channel/queue that temporarily stores messages sent by V8 to
// the front-end, before they are sent over the websocket.
outbound_queue_tx:
UnboundedSender<v8::UniquePtr<v8::inspector::StringBuffer>>,
websocket_tx: WebSocketProxySender,
websocket_rx_handler: Pin<Box<dyn Future<Output = ()> + 'static>>,
}
impl Deref for DenoInspectorSession {
@ -634,7 +696,7 @@ impl DenoInspectorSession {
pub fn new(
inspector_ptr: *mut DenoInspector,
websocket: WebSocket,
websocket: WebSocketProxy,
) -> Box<Self> {
new_box_with(move |self_ptr| {
let v8_channel = v8::inspector::ChannelBase::new::<Self>();
@ -648,54 +710,38 @@ impl DenoInspectorSession {
&empty_view,
);
let (outbound_queue_tx, outbound_queue_rx) =
mpsc::unbounded::<v8::UniquePtr<v8::inspector::StringBuffer>>();
let message_handler =
Self::create_message_handler(self_ptr, websocket, outbound_queue_rx);
let (websocket_tx, websocket_rx) = websocket.split();
let websocket_rx_handler =
Self::receive_from_websocket(self_ptr, websocket_rx);
Self {
v8_channel,
v8_session,
message_handler,
outbound_queue_tx,
websocket_tx,
websocket_rx_handler,
}
})
}
fn create_message_handler(
/// Returns a future that receives messages from the websocket and dispatches
/// them to the V8 session.
fn receive_from_websocket(
self_ptr: *mut Self,
websocket: WebSocket,
outbound_queue_rx: UnboundedReceiver<
v8::UniquePtr<v8::inspector::StringBuffer>,
>,
websocket_rx: WebSocketProxyReceiver,
) -> Pin<Box<dyn Future<Output = ()> + 'static>> {
let (websocket_tx, websocket_rx) = websocket.split();
async move {
eprintln!("Debugger session started.");
// Receive messages from the websocket and dispatch them to the V8 session.
let inbound_pump = websocket_rx
let result = websocket_rx
.map_ok(move |msg| {
let msg = msg.as_bytes();
let msg = v8::inspector::StringView::from(msg);
unsafe { &mut *self_ptr }.dispatch_protocol_message(&msg);
})
.try_collect::<()>();
.try_collect::<()>()
.await;
// Convert and forward messages from the outbound message queue to the
// websocket.
let outbound_pump = outbound_queue_rx
.map(move |msg| {
let msg = msg.unwrap().string().to_string();
let msg = ws::Message::text(msg);
Ok(msg)
})
.forward(websocket_tx);
let disconnect_future = future::try_join(inbound_pump, outbound_pump);
async move {
eprintln!("Debugger session started.");
match disconnect_future.await {
match result {
Ok(_) => eprintln!("Debugger session ended."),
Err(err) => eprintln!("Debugger session ended: {}.", err),
};
@ -703,6 +749,12 @@ impl DenoInspectorSession {
.boxed_local()
}
fn send_to_websocket(&self, msg: v8::UniquePtr<v8::inspector::StringBuffer>) {
let msg = msg.unwrap().string().to_string();
let msg = ws::Message::text(msg);
let _ = self.websocket_tx.unbounded_send(msg);
}
pub fn break_on_first_statement(&mut self) {
let reason = v8::inspector::StringView::from(&b"debugCommand"[..]);
let detail = v8::inspector::StringView::empty();
@ -724,14 +776,14 @@ impl v8::inspector::ChannelImpl for DenoInspectorSession {
_call_id: i32,
message: v8::UniquePtr<v8::inspector::StringBuffer>,
) {
let _ = self.outbound_queue_tx.unbounded_send(message);
self.send_to_websocket(message);
}
fn send_notification(
&mut self,
message: v8::UniquePtr<v8::inspector::StringBuffer>,
) {
let _ = self.outbound_queue_tx.unbounded_send(message);
self.send_to_websocket(message);
}
fn flush_protocol_notifications(&mut self) {}
@ -740,7 +792,7 @@ impl v8::inspector::ChannelImpl for DenoInspectorSession {
impl Future for DenoInspectorSession {
type Output = ();
fn poll(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> {
self.message_handler.poll_unpin(cx)
self.websocket_rx_handler.poll_unpin(cx)
}
}

6
cli/tests/inspector3.js Normal file
View file

@ -0,0 +1,6 @@
for (let i = 0; i < 128; i++) {
console.log(i);
debugger;
}
await new Promise((res, _) => setTimeout(res, 100));
console.log("done");

117
cli/tests/integration_tests.rs
View file

@ -2152,10 +2152,7 @@ fn extract_ws_url_from_stderr(
#[tokio::test]
async fn inspector_connect() {
let script = deno::test_util::root_path()
.join("cli")
.join("tests")
.join("inspector1.js");
let script = util::tests_path().join("inspector1.js");
let mut child = util::deno_cmd()
.arg("run")
// Warning: each inspector test should be on its own port to avoid
@ -2174,6 +2171,7 @@ async fn inspector_connect() {
.expect("Can't connect");
assert_eq!("101 Switching Protocols", response.status().to_string());
child.kill().unwrap();
child.wait().unwrap();
}
enum TestStep {
@ -2184,10 +2182,7 @@ enum TestStep {
#[tokio::test]
async fn inspector_break_on_first_line() {
let script = deno::test_util::root_path()
.join("cli")
.join("tests")
.join("inspector2.js");
let script = util::tests_path().join("inspector2.js");
let mut child = util::deno_cmd()
.arg("run")
// Warning: each inspector test should be on its own port to avoid
@ -2224,12 +2219,12 @@ async fn inspector_break_on_first_line() {
WsRecv(r#"{"id":3,"result":{}}"#),
WsRecv(r#"{"method":"Debugger.paused","#),
WsSend(
r#"{"id":5,"method":"Runtime.evaluate","params":{"expression":"Deno.core.print(\"hello from the inspector\\n\")","contextId":1,"includeCommandLineAPI":true,"silent":false,"returnByValue":true}}"#,
r#"{"id":4,"method":"Runtime.evaluate","params":{"expression":"Deno.core.print(\"hello from the inspector\\n\")","contextId":1,"includeCommandLineAPI":true,"silent":false,"returnByValue":true}}"#,
),
WsRecv(r#"{"id":5,"result":{"result":{"type":"undefined"}}}"#),
WsRecv(r#"{"id":4,"result":{"result":{"type":"undefined"}}}"#),
StdOut("hello from the inspector"),
WsSend(r#"{"id":6,"method":"Debugger.resume"}"#),
WsRecv(r#"{"id":6,"result":{}}"#),
WsSend(r#"{"id":5,"method":"Debugger.resume"}"#),
WsRecv(r#"{"id":5,"result":{}}"#),
StdOut("hello from the script"),
];
@ -2254,14 +2249,12 @@ async fn inspector_break_on_first_line() {
}
child.kill().unwrap();
child.wait().unwrap();
}
#[tokio::test]
async fn inspector_pause() {
let script = deno::test_util::root_path()
.join("cli")
.join("tests")
.join("inspector1.js");
let script = util::tests_path().join("inspector1.js");
let mut child = util::deno_cmd()
.arg("run")
// Warning: each inspector test should be on its own port to avoid
@ -2317,10 +2310,7 @@ async fn inspector_pause() {
#[tokio::test]
async fn inspector_port_collision() {
let script = deno::test_util::root_path()
.join("cli")
.join("tests")
.join("inspector1.js");
let script = util::tests_path().join("inspector1.js");
let mut child1 = util::deno_cmd()
.arg("run")
.arg("--inspect=127.0.0.1:9231")
@ -2348,8 +2338,93 @@ async fn inspector_port_collision() {
.read_to_string(&mut stderr_str_2)
.unwrap();
assert!(stderr_str_2.contains("Cannot start inspector server"));
child1.kill().unwrap();
let _ = child2.kill();
child1.wait().unwrap();
child2.wait().unwrap();
}
#[tokio::test]
async fn inspector_does_not_hang() {
let script = util::tests_path().join("inspector3.js");
let mut child = util::deno_cmd()
.arg("run")
// Warning: each inspector test should be on its own port to avoid
// conflicting with another inspector test.
.arg("--inspect-brk=127.0.0.1:9232")
.arg(script)
.stdout(std::process::Stdio::piped())
.stderr(std::process::Stdio::piped())
.spawn()
.unwrap();
let stderr = child.stderr.as_mut().unwrap();
let ws_url = extract_ws_url_from_stderr(stderr);
let (socket, response) = tokio_tungstenite::connect_async(ws_url)
.await
.expect("Can't connect");
assert_eq!(response.status(), 101); // Switching protocols.
let (mut socket_tx, socket_rx) = socket.split();
let mut socket_rx =
socket_rx.map(|msg| msg.unwrap().to_string()).filter(|msg| {
let pass = !msg.starts_with(r#"{"method":"Debugger.scriptParsed","#);
futures::future::ready(pass)
});
let stdout = child.stdout.as_mut().unwrap();
let mut stdout_lines =
std::io::BufReader::new(stdout).lines().map(|r| r.unwrap());
use TestStep::*;
let test_steps = vec![
WsSend(r#"{"id":1,"method":"Runtime.enable"}"#),
WsSend(r#"{"id":2,"method":"Debugger.enable"}"#),
WsRecv(
r#"{"method":"Runtime.executionContextCreated","params":{"context":{"id":1,"#,
),
WsRecv(r#"{"id":1,"result":{}}"#),
WsRecv(r#"{"id":2,"result":{"debuggerId":"#),
WsSend(r#"{"id":3,"method":"Runtime.runIfWaitingForDebugger"}"#),
WsRecv(r#"{"id":3,"result":{}}"#),
WsRecv(r#"{"method":"Debugger.paused","#),
WsSend(r#"{"id":4,"method":"Debugger.resume"}"#),
WsRecv(r#"{"id":4,"result":{}}"#),
WsRecv(r#"{"method":"Debugger.resumed","params":{}}"#),
];
for step in test_steps {
match step {
WsRecv(s) => assert!(socket_rx.next().await.unwrap().starts_with(s)),
WsSend(s) => socket_tx.send(s.into()).await.unwrap(),
_ => unreachable!(),
}
}
for i in 0..128u32 {
let request_id = i + 10;
// Expect the number {i} on stdout.
let s = format!("{}", i);
assert_eq!(stdout_lines.next().unwrap(), s);
// Expect hitting the `debugger` statement.
let s = r#"{"method":"Debugger.paused","#;
assert!(socket_rx.next().await.unwrap().starts_with(s));
// Send the 'Debugger.resume' request.
let s = format!(r#"{{"id":{},"method":"Debugger.resume"}}"#, request_id);
socket_tx.send(s.into()).await.unwrap();
// Expect confirmation of the 'Debugger.resume' request.
let s = format!(r#"{{"id":{},"result":{{}}}}"#, request_id);
assert_eq!(socket_rx.next().await.unwrap(), s);
let s = r#"{"method":"Debugger.resumed","params":{}}"#;
assert_eq!(socket_rx.next().await.unwrap(), s);
}
// Check that we can gracefully close the websocket connection.
socket_tx.close().await.unwrap();
socket_rx.for_each(|_| async {}).await;
assert_eq!(&stdout_lines.next().unwrap(), "done");
assert!(child.wait().unwrap().success());
}
mod util {