diff options
| -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 | ||||
| -rw-r--r-- | js/dispatch.ts | 74 | ||||
| -rw-r--r-- | js/dispatch_minimal.ts | 77 | ||||
| -rw-r--r-- | js/files.ts | 20 |
7 files changed, 327 insertions, 29 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, ()> { diff --git a/js/dispatch.ts b/js/dispatch.ts index d5e10ad09..64bdfda17 100644 --- a/js/dispatch.ts +++ b/js/dispatch.ts @@ -4,25 +4,53 @@ import * as flatbuffers from "./flatbuffers"; import * as msg from "gen/cli/msg_generated"; import * as errors from "./errors"; import * as util from "./util"; +import { + nextPromiseId, + recordFromBufMinimal, + handleAsyncMsgFromRustMinimal +} from "./dispatch_minimal"; -let nextCmdId = 0; const promiseTable = new Map<number, util.Resolvable<msg.Base>>(); -export function handleAsyncMsgFromRust(ui8: Uint8Array): void { - const bb = new flatbuffers.ByteBuffer(ui8); +interface FlatbufferRecord { + promiseId: number; + base: msg.Base; +} + +function flatbufferRecordFromBuf(buf: Uint8Array): FlatbufferRecord { + const bb = new flatbuffers.ByteBuffer(buf); const base = msg.Base.getRootAsBase(bb); - const cmdId = base.cmdId(); - const promise = promiseTable.get(cmdId); - util.assert(promise != null, `Expecting promise in table. ${cmdId}`); - promiseTable.delete(cmdId); - const err = errors.maybeError(base); - if (err != null) { - promise!.reject(err); + return { + promiseId: base.cmdId(), + base + }; +} + +export function handleAsyncMsgFromRust(ui8: Uint8Array): void { + const buf32 = new Int32Array(ui8.buffer, ui8.byteOffset, ui8.byteLength / 4); + const recordMin = recordFromBufMinimal(buf32); + if (recordMin) { + // Fast and new + handleAsyncMsgFromRustMinimal(ui8, recordMin); } else { - promise!.resolve(base); + // Legacy + let { promiseId, base } = flatbufferRecordFromBuf(ui8); + const promise = promiseTable.get(promiseId); + util.assert(promise != null, `Expecting promise in table. ${promiseId}`); + promiseTable.delete(promiseId); + const err = errors.maybeError(base); + if (err != null) { + promise!.reject(err); + } else { + promise!.resolve(base); + } } } +function ui8FromArrayBufferView(abv: ArrayBufferView): Uint8Array { + return new Uint8Array(abv.buffer, abv.byteOffset, abv.byteLength); +} + function sendInternal( builder: flatbuffers.Builder, innerType: msg.Any, @@ -30,20 +58,20 @@ function sendInternal( zeroCopy: undefined | ArrayBufferView, sync = true ): [number, null | Uint8Array] { - const cmdId = nextCmdId++; - const message = msg.Base.createBase( - builder, - cmdId, - sync, - 0, - 0, - innerType, - inner - ); - builder.finish(message); + const cmdId = nextPromiseId(); + msg.Base.startBase(builder); + msg.Base.addInner(builder, inner); + msg.Base.addInnerType(builder, innerType); + msg.Base.addSync(builder, sync); + msg.Base.addCmdId(builder, cmdId); + builder.finish(msg.Base.endBase(builder)); const control = builder.asUint8Array(); - const response = core.dispatch(control, zeroCopy); + + const response = core.dispatch( + control, + zeroCopy ? ui8FromArrayBufferView(zeroCopy) : undefined + ); builder.inUse = false; return [cmdId, response]; diff --git a/js/dispatch_minimal.ts b/js/dispatch_minimal.ts new file mode 100644 index 000000000..17d328110 --- /dev/null +++ b/js/dispatch_minimal.ts @@ -0,0 +1,77 @@ +// Copyright 2018-2019 the Deno authors. All rights reserved. MIT license. +// Do not add flatbuffer dependencies to this module. +import * as util from "./util"; +import { core } from "./core"; + +const DISPATCH_MINIMAL_TOKEN = 0xcafe; +const promiseTableMin = new Map<number, util.Resolvable<number>>(); +let _nextPromiseId = 0; + +export function nextPromiseId(): number { + return _nextPromiseId++; +} + +export interface RecordMinimal { + promiseId: number; + opId: number; + arg: number; + result: number; +} + +/** Determines if a message has the "minimal" serialization format. If false, it + * is flatbuffer encoded. + */ +export function hasMinimalToken(i32: Int32Array): boolean { + return i32[0] == DISPATCH_MINIMAL_TOKEN; +} + +export function recordFromBufMinimal(buf32: Int32Array): null | RecordMinimal { + if (hasMinimalToken(buf32)) { + return { + promiseId: buf32[1], + opId: buf32[2], + arg: buf32[3], + result: buf32[4] + }; + } + return null; +} + +const scratch32 = new Int32Array(5); +const scratchBytes = new Uint8Array( + scratch32.buffer, + scratch32.byteOffset, + scratch32.byteLength +); +util.assert(scratchBytes.byteLength === scratch32.length * 4); + +export function handleAsyncMsgFromRustMinimal( + ui8: Uint8Array, + record: RecordMinimal +): void { + // Fast and new + util.log("minimal handleAsyncMsgFromRust ", ui8.length); + const { promiseId, result } = record; + const promise = promiseTableMin.get(promiseId); + promiseTableMin.delete(promiseId); + promise!.resolve(result); +} + +export function sendAsyncMinimal( + opId: number, + arg: number, + zeroCopy: Uint8Array +): Promise<number> { + const promiseId = nextPromiseId(); // AKA cmdId + + scratch32[0] = DISPATCH_MINIMAL_TOKEN; + scratch32[1] = promiseId; + scratch32[2] = opId; + scratch32[3] = arg; + + const promise = util.createResolvable<number>(); + promiseTableMin.set(promiseId, promise); + + core.dispatch(scratchBytes, zeroCopy); + return promise; +} diff --git a/js/files.ts b/js/files.ts index 18ee612bc..1479e4146 100644 --- a/js/files.ts +++ b/js/files.ts @@ -11,10 +11,14 @@ import { SyncSeeker } from "./io"; import * as dispatch from "./dispatch"; +import { sendAsyncMinimal } from "./dispatch_minimal"; import * as msg from "gen/cli/msg_generated"; import { assert } from "./util"; import * as flatbuffers from "./flatbuffers"; +const OP_READ = 1; +const OP_WRITE = 2; + function reqOpen( filename: string, mode: OpenMode @@ -101,7 +105,14 @@ export function readSync(rid: number, p: Uint8Array): ReadResult { * })(); */ export async function read(rid: number, p: Uint8Array): Promise<ReadResult> { - return resRead(await dispatch.sendAsync(...reqRead(rid, p))); + const nread = await sendAsyncMinimal(OP_READ, rid, p); + if (nread < 0) { + throw new Error("read error"); + } else if (nread == 0) { + return { nread, eof: true }; + } else { + return { nread, eof: false }; + } } function reqWrite( @@ -147,7 +158,12 @@ export function writeSync(rid: number, p: Uint8Array): number { * */ export async function write(rid: number, p: Uint8Array): Promise<number> { - return resWrite(await dispatch.sendAsync(...reqWrite(rid, p))); + let result = await sendAsyncMinimal(OP_WRITE, rid, p); + if (result < 0) { + throw new Error("write error"); + } else { + return result; + } } function reqSeek( |