/// To run this benchmark: /// /// > DENO_BUILD_MODE=release ./tools/build.py && \ /// ./target/release/deno_core_http_bench --multi-thread extern crate deno_core; extern crate futures; extern crate libc; extern crate tokio; #[macro_use] extern crate log; #[macro_use] extern crate lazy_static; use deno_core::*; use futures::future::lazy; use std::collections::HashMap; use std::env; use std::mem; use std::net::SocketAddr; use std::sync::atomic::AtomicUsize; use std::sync::atomic::Ordering; use std::sync::Mutex; use tokio::prelude::*; const OP_LISTEN: i32 = 1; const OP_ACCEPT: i32 = 2; const OP_READ: i32 = 3; const OP_WRITE: i32 = 4; const OP_CLOSE: i32 = 5; const INDEX_START: usize = 0; const INDEX_END: usize = 1; const NUM_RECORDS: usize = 128; const RECORD_SIZE: usize = 4; #[derive(Clone, Debug)] pub struct Record { pub promise_id: i32, pub op_id: i32, pub arg: i32, pub result: i32, } pub type HttpBenchOp = dyn Future + Send; struct HttpBench { shared32: Vec, } impl HttpBench { fn new() -> Self { let mut shared32 = Vec::::new(); let n = 2 + 4 * NUM_RECORDS; shared32.resize(n, 0); shared32[INDEX_START] = 0; shared32[INDEX_END] = 0; Self { shared32 } } } fn idx(i: usize, off: usize) -> usize { 2 + i * RECORD_SIZE + off } impl Behavior for HttpBench { fn startup_snapshot(&mut self) -> Option { None } fn startup_shared(&mut self) -> Option { let ptr = self.shared32.as_ptr() as *const u8; let len = mem::size_of::() * self.shared32.len(); Some(unsafe { deno_buf::from_raw_parts(ptr, len) }) } fn resolve(&mut self, _specifier: &str, _referrer: deno_mod) -> deno_mod { // HttpBench doesn't do ES modules. unimplemented!() } fn recv( &mut self, record: Record, zero_copy_buf: deno_buf, ) -> (bool, Box>) { let is_sync = record.promise_id == 0; let http_bench_op = match record.op_id { OP_LISTEN => { assert!(is_sync); op_listen() } OP_CLOSE => { assert!(is_sync); let rid = record.arg; op_close(rid) } OP_ACCEPT => { assert!(!is_sync); let listener_rid = record.arg; op_accept(listener_rid) } OP_READ => { assert!(!is_sync); let rid = record.arg; op_read(rid, zero_copy_buf) } OP_WRITE => { assert!(!is_sync); let rid = record.arg; op_write(rid, zero_copy_buf) } _ => panic!("bad op {}", record.op_id), }; let mut record_a = record.clone(); let mut record_b = record.clone(); let op = Box::new( http_bench_op .and_then(move |result| { record_a.result = result; Ok(record_a) }).or_else(|err| -> Result { eprintln!("unexpected err {}", err); record_b.result = -1; Ok(record_b) }), ); (is_sync, op) } fn records_reset(&mut self) { self.shared32[INDEX_START] = 0; self.shared32[INDEX_END] = 0; } fn records_push(&mut self, record: Record) -> bool { debug!("push {:?}", record); let i = self.shared32[INDEX_END] as usize; if i >= NUM_RECORDS { return false; } self.shared32[idx(i, 0)] = record.promise_id; self.shared32[idx(i, 1)] = record.op_id; self.shared32[idx(i, 2)] = record.arg; self.shared32[idx(i, 3)] = record.result; self.shared32[INDEX_END] += 1; true } fn records_shift(&mut self) -> Option { let i = self.shared32[INDEX_START] as usize; if i == self.shared32[INDEX_END] as usize { return None; } let record = Record { promise_id: self.shared32[idx(i, 0)], op_id: self.shared32[idx(i, 1)], arg: self.shared32[idx(i, 2)], result: self.shared32[idx(i, 3)], }; self.shared32[INDEX_START] += 1; Some(record) } } fn main() { let js_source = include_str!("http_bench.js"); let main_future = lazy(move || { let isolate = deno_core::Isolate::new(HttpBench::new()); // TODO currently isolate.execute() must be run inside tokio, hence the // lazy(). It would be nice to not have that contraint. Probably requires // using v8::MicrotasksPolicy::kExplicit js_check(isolate.execute("http_bench.js", js_source)); isolate.then(|r| { js_check(r); Ok(()) }) }); let args: Vec = env::args().collect(); if args.len() > 1 && args[1] == "--multi-thread" { println!("multi-thread"); tokio::run(main_future); } else { println!("single-thread"); tokio::runtime::current_thread::run(main_future); } } enum Repr { TcpListener(tokio::net::TcpListener), TcpStream(tokio::net::TcpStream), } type ResourceTable = HashMap; lazy_static! { static ref RESOURCE_TABLE: Mutex = Mutex::new(HashMap::new()); static ref NEXT_RID: AtomicUsize = AtomicUsize::new(3); } fn new_rid() -> i32 { let rid = NEXT_RID.fetch_add(1, Ordering::SeqCst); rid as i32 } fn op_accept(listener_rid: i32) -> Box { debug!("accept {}", listener_rid); Box::new( futures::future::poll_fn(move || { let mut table = RESOURCE_TABLE.lock().unwrap(); let maybe_repr = table.get_mut(&listener_rid); match maybe_repr { Some(Repr::TcpListener(ref mut listener)) => listener.poll_accept(), _ => panic!("bad rid {}", listener_rid), } }).and_then(move |(stream, addr)| { debug!("accept success {}", addr); let rid = new_rid(); let mut guard = RESOURCE_TABLE.lock().unwrap(); guard.insert(rid, Repr::TcpStream(stream)); Ok(rid as i32) }), ) } fn op_listen() -> Box { debug!("listen"); Box::new(lazy(move || { let addr = "127.0.0.1:4544".parse::().unwrap(); let listener = tokio::net::TcpListener::bind(&addr).unwrap(); let rid = new_rid(); let mut guard = RESOURCE_TABLE.lock().unwrap(); guard.insert(rid, Repr::TcpListener(listener)); futures::future::ok(rid) })) } fn op_close(rid: i32) -> Box { debug!("close"); Box::new(lazy(move || { let mut table = RESOURCE_TABLE.lock().unwrap(); let r = table.remove(&rid); let result = if r.is_some() { 0 } else { -1 }; futures::future::ok(result) })) } fn op_read(rid: i32, mut zero_copy_buf: deno_buf) -> Box { debug!("read rid={}", rid); Box::new( futures::future::poll_fn(move || { let mut table = RESOURCE_TABLE.lock().unwrap(); let maybe_repr = table.get_mut(&rid); match maybe_repr { Some(Repr::TcpStream(ref mut stream)) => { stream.poll_read(&mut zero_copy_buf) } _ => panic!("bad rid"), } }).and_then(move |nread| { debug!("read success {}", nread); Ok(nread as i32) }), ) } fn op_write(rid: i32, zero_copy_buf: deno_buf) -> Box { debug!("write rid={}", rid); Box::new( futures::future::poll_fn(move || { let mut table = RESOURCE_TABLE.lock().unwrap(); let maybe_repr = table.get_mut(&rid); match maybe_repr { Some(Repr::TcpStream(ref mut stream)) => { stream.poll_write(&zero_copy_buf) } _ => panic!("bad rid"), } }).and_then(move |nwritten| { debug!("write success {}", nwritten); Ok(nwritten as i32) }), ) } fn js_check(r: Result<(), JSError>) { if let Err(e) = r { panic!(e.to_string()); } }