diff options
Diffstat (limited to 'cli')
| -rw-r--r-- | cli/BUILD.gn | 1 | ||||
| -rw-r--r-- | cli/dispatch_minimal.rs | 159 | ||||
| -rw-r--r-- | cli/main.rs | 1 | ||||
| -rw-r--r-- | cli/ops.rs | 24 |
4 files changed, 181 insertions, 4 deletions
diff --git a/cli/BUILD.gn b/cli/BUILD.gn index 936b0592e..0b5fadc68 100644 --- a/cli/BUILD.gn +++ b/cli/BUILD.gn @@ -64,6 +64,7 @@ ts_sources = [ "../js/deno.ts", "../js/dir.ts", "../js/dispatch.ts", + "../js/dispatch_minimal.ts", "../js/dom_types.ts", "../js/errors.ts", "../js/event.ts", diff --git a/cli/dispatch_minimal.rs b/cli/dispatch_minimal.rs new file mode 100644 index 000000000..a3a383393 --- /dev/null +++ b/cli/dispatch_minimal.rs @@ -0,0 +1,159 @@ +// Copyright 2018-2019 the Deno authors. All rights reserved. MIT license. +// Do not add flatbuffer dependencies to this module. +//! Connects to js/dispatch_minimal.ts sendAsyncMinimal This acts as a faster +//! alternative to flatbuffers using a very simple list of int32s to lay out +//! messages. The first i32 is used to determine if a message a flatbuffer +//! message or a "minimal" message. +use crate::state::ThreadSafeState; +use deno::Buf; +use deno::Op; +use deno::PinnedBuf; +use futures::Future; + +const DISPATCH_MINIMAL_TOKEN: i32 = 0xCAFE; +const OP_READ: i32 = 1; +const OP_WRITE: i32 = 2; + +#[derive(Copy, Clone, Debug, PartialEq)] +// This corresponds to RecordMinimal on the TS side. +pub struct Record { + pub promise_id: i32, + pub op_id: i32, + pub arg: i32, + pub result: i32, +} + +impl Into<Buf> for Record { + fn into(self) -> Buf { + let vec = vec![ + DISPATCH_MINIMAL_TOKEN, + self.promise_id, + self.op_id, + self.arg, + self.result, + ]; + let buf32 = vec.into_boxed_slice(); + let ptr = Box::into_raw(buf32) as *mut [u8; 5 * 4]; + unsafe { Box::from_raw(ptr) } + } +} + +pub fn parse_min_record(bytes: &[u8]) -> Option<Record> { + if bytes.len() % std::mem::size_of::<i32>() != 0 { + return None; + } + let p = bytes.as_ptr(); + #[allow(clippy::cast_ptr_alignment)] + let p32 = p as *const i32; + let s = unsafe { std::slice::from_raw_parts(p32, bytes.len() / 4) }; + + if s.len() < 5 { + return None; + } + let ptr = s.as_ptr(); + let ints = unsafe { std::slice::from_raw_parts(ptr, 5) }; + if ints[0] != DISPATCH_MINIMAL_TOKEN { + return None; + } + Some(Record { + promise_id: ints[1], + op_id: ints[2], + arg: ints[3], + result: ints[4], + }) +} + +#[test] +fn test_parse_min_record() { + let buf = vec![ + 0xFE, 0xCA, 0, 0, 1, 0, 0, 0, 2, 0, 0, 0, 3, 0, 0, 0, 4, 0, 0, 0, + ]; + assert_eq!( + parse_min_record(&buf), + Some(Record { + promise_id: 1, + op_id: 2, + arg: 3, + result: 4, + }) + ); + + let buf = vec![]; + assert_eq!(parse_min_record(&buf), None); + + let buf = vec![5]; + assert_eq!(parse_min_record(&buf), None); +} + +pub fn dispatch_minimal( + state: &ThreadSafeState, + mut record: Record, + zero_copy: Option<PinnedBuf>, +) -> Op { + let is_sync = record.promise_id == 0; + let min_op = match record.op_id { + OP_READ => ops::read(record.arg, zero_copy), + OP_WRITE => ops::write(record.arg, zero_copy), + _ => unimplemented!(), + }; + + let state = state.clone(); + + let fut = Box::new(min_op.then(move |result| -> Result<Buf, ()> { + match result { + Ok(r) => { + record.result = r; + } + Err(err) => { + // TODO(ry) The dispatch_minimal doesn't properly pipe errors back to + // the caller. + debug!("swallowed err {}", err); + record.result = -1; + } + } + let buf: Buf = record.into(); + state.metrics_op_completed(buf.len()); + Ok(buf) + })); + if is_sync { + Op::Sync(fut.wait().unwrap()) + } else { + Op::Async(fut) + } +} + +mod ops { + use crate::errors; + use crate::resources; + use crate::tokio_write; + use deno::PinnedBuf; + use futures::Future; + + type MinimalOp = dyn Future<Item = i32, Error = errors::DenoError> + Send; + + pub fn read(rid: i32, zero_copy: Option<PinnedBuf>) -> Box<MinimalOp> { + debug!("read rid={}", rid); + let zero_copy = zero_copy.unwrap(); + match resources::lookup(rid as u32) { + None => Box::new(futures::future::err(errors::bad_resource())), + Some(resource) => Box::new( + tokio::io::read(resource, zero_copy) + .map_err(errors::DenoError::from) + .and_then(move |(_resource, _buf, nread)| Ok(nread as i32)), + ), + } + } + + pub fn write(rid: i32, zero_copy: Option<PinnedBuf>) -> Box<MinimalOp> { + debug!("write rid={}", rid); + let zero_copy = zero_copy.unwrap(); + match resources::lookup(rid as u32) { + None => Box::new(futures::future::err(errors::bad_resource())), + Some(resource) => Box::new( + tokio_write::write(resource, zero_copy) + .map_err(errors::DenoError::from) + .and_then(move |(_resource, _buf, nwritten)| Ok(nwritten as i32)), + ), + } + } +} diff --git a/cli/main.rs b/cli/main.rs index c601c00b1..45ab334a9 100644 --- a/cli/main.rs +++ b/cli/main.rs @@ -15,6 +15,7 @@ extern crate nix; mod ansi; pub mod compiler; pub mod deno_dir; +mod dispatch_minimal; pub mod errors; pub mod flags; mod fs; diff --git a/cli/ops.rs b/cli/ops.rs index 4c644714f..b75140dcc 100644 --- a/cli/ops.rs +++ b/cli/ops.rs @@ -2,6 +2,8 @@ use atty; use crate::ansi; use crate::compiler::get_compiler_config; +use crate::dispatch_minimal::dispatch_minimal; +use crate::dispatch_minimal::parse_min_record; use crate::errors; use crate::errors::{DenoError, DenoResult, ErrorKind}; use crate::fs as deno_fs; @@ -74,18 +76,33 @@ fn empty_buf() -> Buf { Box::new([]) } +pub fn dispatch_all( + state: &ThreadSafeState, + control: &[u8], + zero_copy: Option<PinnedBuf>, + op_selector: OpSelector, +) -> Op { + let bytes_sent_control = control.len(); + let bytes_sent_zero_copy = zero_copy.as_ref().map(|b| b.len()).unwrap_or(0); + let op = if let Some(min_record) = parse_min_record(control) { + dispatch_minimal(state, min_record, zero_copy) + } else { + dispatch_all_legacy(state, control, zero_copy, op_selector) + }; + state.metrics_op_dispatched(bytes_sent_control, bytes_sent_zero_copy); + op +} + /// Processes raw messages from JavaScript. /// This functions invoked every time Deno.core.dispatch() is called. /// control corresponds to the first argument of Deno.core.dispatch(). /// data corresponds to the second argument of Deno.core.dispatch(). -pub fn dispatch_all( +pub fn dispatch_all_legacy( state: &ThreadSafeState, control: &[u8], zero_copy: Option<PinnedBuf>, op_selector: OpSelector, ) -> Op { - let bytes_sent_control = control.len(); - let bytes_sent_zero_copy = zero_copy.as_ref().map(|b| b.len()).unwrap_or(0); let base = msg::get_root_as_base(&control); let is_sync = base.sync(); let inner_type = base.inner_type(); @@ -99,7 +116,6 @@ pub fn dispatch_all( let op: Box<OpWithError> = op_func(state, &base, zero_copy); let state = state.clone(); - state.metrics_op_dispatched(bytes_sent_control, bytes_sent_zero_copy); let fut = Box::new( op.or_else(move |err: DenoError| -> Result<Buf, ()> { |