diff options
Diffstat (limited to 'core/examples/http_bench_bin_ops.rs')
-rw-r--r-- | core/examples/http_bench_bin_ops.rs | 223 |
1 files changed, 105 insertions, 118 deletions
diff --git a/core/examples/http_bench_bin_ops.rs b/core/examples/http_bench_bin_ops.rs index 366779e8c..f93b8b079 100644 --- a/core/examples/http_bench_bin_ops.rs +++ b/core/examples/http_bench_bin_ops.rs @@ -1,10 +1,12 @@ #[macro_use] extern crate log; +use deno_core::js_check; +use deno_core::BasicState; +use deno_core::BufVec; use deno_core::CoreIsolate; -use deno_core::CoreIsolateState; use deno_core::Op; -use deno_core::ResourceTable; +use deno_core::OpRegistry; use deno_core::Script; use deno_core::StartupData; use deno_core::ZeroCopyBuf; @@ -12,7 +14,6 @@ use futures::future::poll_fn; use futures::future::FutureExt; use futures::future::TryFuture; use futures::future::TryFutureExt; -use std::cell::RefCell; use std::convert::TryInto; use std::env; use std::fmt::Debug; @@ -27,6 +28,7 @@ use tokio::io::AsyncRead; use tokio::io::AsyncWrite; use tokio::net::TcpListener; use tokio::net::TcpStream; +use tokio::runtime; struct Logger; @@ -46,9 +48,9 @@ impl log::Log for Logger { #[derive(Copy, Clone, Debug, PartialEq)] struct Record { - pub promise_id: u32, - pub rid: u32, - pub result: i32, + promise_id: u32, + rid: u32, + result: i32, } type RecordBuf = [u8; size_of::<Record>()]; @@ -75,131 +77,64 @@ impl From<Record> for RecordBuf { } } -pub fn isolate_new() -> CoreIsolate { +fn create_isolate() -> CoreIsolate { + let state = BasicState::new(); + register_op_bin_sync(&state, "listen", op_listen); + register_op_bin_sync(&state, "close", op_close); + register_op_bin_async(&state, "accept", op_accept); + register_op_bin_async(&state, "read", op_read); + register_op_bin_async(&state, "write", op_write); + let startup_data = StartupData::Script(Script { source: include_str!("http_bench_bin_ops.js"), filename: "http_bench_bin_ops.js", }); - let mut isolate = CoreIsolate::new(startup_data, false); - - fn register_sync_op<F>( - isolate: &mut CoreIsolate, - name: &'static str, - handler: F, - ) where - F: 'static - + Fn( - Rc<RefCell<ResourceTable>>, - u32, - &mut [ZeroCopyBuf], - ) -> Result<u32, Error>, - { - let core_handler = move |state: &mut CoreIsolateState, - zero_copy_bufs: &mut [ZeroCopyBuf]| - -> Op { - assert!(!zero_copy_bufs.is_empty()); - let record = Record::from(zero_copy_bufs[0].as_ref()); - let is_sync = record.promise_id == 0; - assert!(is_sync); - - let resource_table = state.resource_table.clone(); - let result: i32 = - match handler(resource_table, record.rid, &mut zero_copy_bufs[1..]) { - Ok(r) => r as i32, - Err(_) => -1, - }; - let buf = RecordBuf::from(Record { result, ..record })[..].into(); - Op::Sync(buf) - }; - - isolate.register_op(name, core_handler); - } - - fn register_async_op<F>( - isolate: &mut CoreIsolate, - name: &'static str, - handler: impl Fn(Rc<RefCell<ResourceTable>>, u32, &mut [ZeroCopyBuf]) -> F - + Copy - + 'static, - ) where - F: TryFuture, - F::Ok: TryInto<i32>, - <F::Ok as TryInto<i32>>::Error: Debug, - { - let core_handler = move |state: &mut CoreIsolateState, - zero_copy_bufs: &mut [ZeroCopyBuf]| - -> Op { - assert!(!zero_copy_bufs.is_empty()); - let record = Record::from(zero_copy_bufs[0].as_ref()); - let is_sync = record.promise_id == 0; - assert!(!is_sync); - - let mut zero_copy = zero_copy_bufs[1..].to_vec(); - let resource_table = state.resource_table.clone(); - let fut = async move { - let op = handler(resource_table, record.rid, &mut zero_copy); - let result = op - .map_ok(|r| r.try_into().expect("op result does not fit in i32")) - .unwrap_or_else(|_| -1) - .await; - RecordBuf::from(Record { result, ..record })[..].into() - }; - - Op::Async(fut.boxed_local()) - }; - - isolate.register_op(name, core_handler); - } - - register_sync_op(&mut isolate, "listen", op_listen); - register_async_op(&mut isolate, "accept", op_accept); - register_async_op(&mut isolate, "read", op_read); - register_async_op(&mut isolate, "write", op_write); - register_sync_op(&mut isolate, "close", op_close); - - isolate -} - -fn op_close( - resource_table: Rc<RefCell<ResourceTable>>, - rid: u32, - _buf: &mut [ZeroCopyBuf], -) -> Result<u32, Error> { - debug!("close rid={}", rid); - let resource_table = &mut resource_table.borrow_mut(); - resource_table - .close(rid) - .map(|_| 0) - .ok_or_else(bad_resource) + CoreIsolate::new(state, startup_data, false) } fn op_listen( - resource_table: Rc<RefCell<ResourceTable>>, + state: &BasicState, _rid: u32, - _buf: &mut [ZeroCopyBuf], + _bufs: &mut [ZeroCopyBuf], ) -> Result<u32, Error> { debug!("listen"); let addr = "127.0.0.1:4544".parse::<SocketAddr>().unwrap(); let std_listener = std::net::TcpListener::bind(&addr)?; let listener = TcpListener::from_std(std_listener)?; - let resource_table = &mut resource_table.borrow_mut(); - let rid = resource_table.add("tcpListener", Box::new(listener)); + let rid = state + .resource_table + .borrow_mut() + .add("tcpListener", Box::new(listener)); Ok(rid) } +fn op_close( + state: &BasicState, + rid: u32, + _bufs: &mut [ZeroCopyBuf], +) -> Result<u32, Error> { + debug!("close rid={}", rid); + state + .resource_table + .borrow_mut() + .close(rid) + .map(|_| 0) + .ok_or_else(bad_resource_id) +} + fn op_accept( - resource_table: Rc<RefCell<ResourceTable>>, + state: Rc<BasicState>, rid: u32, - _buf: &mut [ZeroCopyBuf], + _bufs: BufVec, ) -> impl TryFuture<Ok = u32, Error = Error> { debug!("accept rid={}", rid); poll_fn(move |cx| { - let resource_table = &mut resource_table.borrow_mut(); + let resource_table = &mut state.resource_table.borrow_mut(); let listener = resource_table .get_mut::<TcpListener>(rid) - .ok_or_else(bad_resource)?; + .ok_or_else(bad_resource_id)?; listener.poll_accept(cx).map_ok(|(stream, _addr)| { resource_table.add("tcpStream", Box::new(stream)) }) @@ -207,9 +142,9 @@ fn op_accept( } fn op_read( - resource_table: Rc<RefCell<ResourceTable>>, + state: Rc<BasicState>, rid: u32, - bufs: &mut [ZeroCopyBuf], + bufs: BufVec, ) -> impl TryFuture<Ok = usize, Error = Error> { assert_eq!(bufs.len(), 1, "Invalid number of arguments"); let mut buf = bufs[0].clone(); @@ -217,33 +152,85 @@ fn op_read( debug!("read rid={}", rid); poll_fn(move |cx| { - let resource_table = &mut resource_table.borrow_mut(); + let resource_table = &mut state.resource_table.borrow_mut(); let stream = resource_table .get_mut::<TcpStream>(rid) - .ok_or_else(bad_resource)?; + .ok_or_else(bad_resource_id)?; Pin::new(stream).poll_read(cx, &mut buf) }) } fn op_write( - resource_table: Rc<RefCell<ResourceTable>>, + state: Rc<BasicState>, rid: u32, - bufs: &mut [ZeroCopyBuf], + bufs: BufVec, ) -> impl TryFuture<Ok = usize, Error = Error> { assert_eq!(bufs.len(), 1, "Invalid number of arguments"); let buf = bufs[0].clone(); debug!("write rid={}", rid); poll_fn(move |cx| { - let resource_table = &mut resource_table.borrow_mut(); + let resource_table = &mut state.resource_table.borrow_mut(); let stream = resource_table .get_mut::<TcpStream>(rid) - .ok_or_else(bad_resource)?; + .ok_or_else(bad_resource_id)?; Pin::new(stream).poll_write(cx, &buf) }) } -fn bad_resource() -> Error { +fn register_op_bin_sync<F>(state: &BasicState, name: &'static str, op_fn: F) +where + F: Fn(&BasicState, u32, &mut [ZeroCopyBuf]) -> Result<u32, Error> + 'static, +{ + let base_op_fn = move |state: Rc<BasicState>, mut bufs: BufVec| -> Op { + let record = Record::from(bufs[0].as_ref()); + let is_sync = record.promise_id == 0; + assert!(is_sync); + + let zero_copy_bufs = &mut bufs[1..]; + let result: i32 = match op_fn(&state, record.rid, zero_copy_bufs) { + Ok(r) => r as i32, + Err(_) => -1, + }; + let buf = RecordBuf::from(Record { result, ..record })[..].into(); + Op::Sync(buf) + }; + + state.register_op(name, base_op_fn); +} + +fn register_op_bin_async<F, R>(state: &BasicState, name: &'static str, op_fn: F) +where + F: Fn(Rc<BasicState>, u32, BufVec) -> R + Copy + 'static, + R: TryFuture, + R::Ok: TryInto<i32>, + <R::Ok as TryInto<i32>>::Error: Debug, +{ + let base_op_fn = move |state: Rc<BasicState>, bufs: BufVec| -> Op { + let mut bufs_iter = bufs.into_iter(); + let record_buf = bufs_iter.next().unwrap(); + let zero_copy_bufs = bufs_iter.collect::<BufVec>(); + + let record = Record::from(record_buf.as_ref()); + let is_sync = record.promise_id == 0; + assert!(!is_sync); + + let fut = async move { + let op = op_fn(state, record.rid, zero_copy_bufs); + let result = op + .map_ok(|r| r.try_into().expect("op result does not fit in i32")) + .unwrap_or_else(|_| -1) + .await; + RecordBuf::from(Record { result, ..record })[..].into() + }; + + Op::Async(fut.boxed_local()) + }; + + state.register_op(name, base_op_fn); +} + +fn bad_resource_id() -> Error { Error::new(ErrorKind::NotFound, "bad resource id") } @@ -259,13 +246,13 @@ fn main() { // NOTE: `--help` arg will display V8 help and exit deno_core::v8_set_flags(env::args().collect()); - let isolate = isolate_new(); - let mut runtime = tokio::runtime::Builder::new() + let isolate = create_isolate(); + let mut runtime = runtime::Builder::new() .basic_scheduler() .enable_all() .build() .unwrap(); - runtime.block_on(isolate).expect("unexpected isolate error"); + js_check(runtime.block_on(isolate)); } #[test] |