diff --git a/Cargo.lock b/Cargo.lock index 2f9fe45180..4fbd10fcea 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -255,6 +255,7 @@ dependencies = [ name = "deno" version = "0.21.0" dependencies = [ + "downcast-rs 1.1.0 (registry+https://github.com/rust-lang/crates.io-index)", "futures 0.1.29 (registry+https://github.com/rust-lang/crates.io-index)", "lazy_static 1.4.0 (registry+https://github.com/rust-lang/crates.io-index)", "libc 0.2.62 (registry+https://github.com/rust-lang/crates.io-index)", @@ -341,6 +342,11 @@ dependencies = [ "winapi 0.3.8 (registry+https://github.com/rust-lang/crates.io-index)", ] +[[package]] +name = "downcast-rs" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" + [[package]] name = "dtoa" version = "0.4.4" @@ -1990,6 +1996,7 @@ dependencies = [ "checksum ct-logs 0.6.0 (registry+https://github.com/rust-lang/crates.io-index)" = "4d3686f5fa27dbc1d76c751300376e167c5a43387f44bb451fd1c24776e49113" "checksum dirs 2.0.2 (registry+https://github.com/rust-lang/crates.io-index)" = "13aea89a5c93364a98e9b37b2fa237effbb694d5cfe01c5b70941f7eb087d5e3" "checksum dirs-sys 0.3.4 (registry+https://github.com/rust-lang/crates.io-index)" = "afa0b23de8fd801745c471deffa6e12d248f962c9fd4b4c33787b055599bde7b" +"checksum downcast-rs 1.1.0 (registry+https://github.com/rust-lang/crates.io-index)" = "5fe414cc2fd4447b7da94b27ddfb6831a8a06f35f6d077ab5613ec703866c49a" "checksum dtoa 0.4.4 (registry+https://github.com/rust-lang/crates.io-index)" = "ea57b42383d091c85abcc2706240b94ab2a8fa1fc81c10ff23c4de06e2a90b5e" "checksum either 1.5.3 (registry+https://github.com/rust-lang/crates.io-index)" = "bb1f6b1ce1c140482ea30ddd3335fc0024ac7ee112895426e0a629a6c20adfe3" "checksum encoding_rs 0.8.19 (registry+https://github.com/rust-lang/crates.io-index)" = "79906e1ad1f7f8bc48864fcc6ffd58336fb5992e627bf61928099cb25fdf4314" diff --git a/core/Cargo.toml b/core/Cargo.toml index 241e8a2ac6..cbc19edbc0 100644 --- a/core/Cargo.toml +++ b/core/Cargo.toml @@ -14,6 +14,7 @@ repository = "https://github.com/denoland/deno" path = "lib.rs" [dependencies] +downcast-rs = "1.1.0" futures = "0.1.29" lazy_static = "1.4.0" libc = "0.2.62" diff --git a/core/examples/http_bench.rs b/core/examples/http_bench.rs index c019d8a119..a6cc6d5482 100644 --- a/core/examples/http_bench.rs +++ b/core/examples/http_bench.rs @@ -14,12 +14,10 @@ extern crate lazy_static; use deno::*; use futures::future::lazy; -use std::collections::HashMap; use std::env; use std::net::SocketAddr; -use std::sync::atomic::AtomicUsize; -use std::sync::atomic::Ordering; use std::sync::Mutex; +use std::sync::MutexGuard; use tokio::prelude::*; static LOGGER: Logger = Logger; @@ -184,44 +182,38 @@ fn main() { } } -enum Repr { - TcpListener(tokio::net::TcpListener), - TcpStream(tokio::net::TcpStream), -} +struct TcpListener(tokio::net::TcpListener); + +impl Resource for TcpListener {} + +struct TcpStream(tokio::net::TcpStream); + +impl Resource for TcpStream {} -type ResourceTable = HashMap; lazy_static! { - static ref RESOURCE_TABLE: Mutex = Mutex::new(HashMap::new()); - static ref NEXT_RID: AtomicUsize = AtomicUsize::new(3); + static ref RESOURCE_TABLE: Mutex = + Mutex::new(ResourceTable::default()); } -fn new_rid() -> i32 { - let rid = NEXT_RID.fetch_add(1, Ordering::SeqCst); - rid as i32 +fn lock_resource_table<'a>() -> MutexGuard<'a, ResourceTable> { + RESOURCE_TABLE.lock().unwrap() } fn op_accept(record: Record, _zero_copy_buf: Option) -> Box { - let listener_rid = record.arg; - 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) - }), - ) + let rid = record.arg as u32; + debug!("accept {}", rid); + let fut = futures::future::poll_fn(move || { + let mut table = lock_resource_table(); + let listener = table.get_mut::(rid)?; + listener.0.poll_accept() + }) + .and_then(move |(stream, addr)| { + debug!("accept success {}", addr); + let mut table = lock_resource_table(); + let rid = table.add(Box::new(TcpStream(stream))); + Ok(rid as i32) + }); + Box::new(fut) } fn op_listen( @@ -229,70 +221,54 @@ fn op_listen( _zero_copy_buf: Option, ) -> 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) - })) + let addr = "127.0.0.1:4544".parse::().unwrap(); + let listener = tokio::net::TcpListener::bind(&addr).unwrap(); + let mut table = lock_resource_table(); + let rid = table.add(Box::new(TcpListener(listener))); + Box::new(futures::future::ok(rid as i32)) } fn op_close(record: Record, _zero_copy_buf: Option) -> Box { debug!("close"); - let rid = record.arg; - 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) - })) + let rid = record.arg as u32; + let mut table = lock_resource_table(); + let fut = match table.close(rid) { + Ok(_) => futures::future::ok(0), + Err(e) => futures::future::err(e), + }; + Box::new(fut) } fn op_read(record: Record, zero_copy_buf: Option) -> Box { - let rid = record.arg; + let rid = record.arg as u32; debug!("read rid={}", rid); let mut zero_copy_buf = zero_copy_buf.unwrap(); - 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) - }), - ) + let fut = futures::future::poll_fn(move || { + let mut table = lock_resource_table(); + let stream = table.get_mut::(rid)?; + stream.0.poll_read(&mut zero_copy_buf) + }) + .and_then(move |nread| { + debug!("read success {}", nread); + Ok(nread as i32) + }); + Box::new(fut) } fn op_write(record: Record, zero_copy_buf: Option) -> Box { - let rid = record.arg; + let rid = record.arg as u32; debug!("write rid={}", rid); let zero_copy_buf = zero_copy_buf.unwrap(); - 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) - }), - ) + let fut = futures::future::poll_fn(move || { + let mut table = lock_resource_table(); + let stream = table.get_mut::(rid)?; + stream.0.poll_write(&zero_copy_buf) + }) + .and_then(move |nwritten| { + debug!("write success {}", nwritten); + Ok(nwritten as i32) + }); + Box::new(fut) } fn js_check(r: Result<(), ErrBox>) { diff --git a/core/lib.rs b/core/lib.rs index 42a692f1a1..31f717769e 100644 --- a/core/lib.rs +++ b/core/lib.rs @@ -3,6 +3,8 @@ extern crate log; extern crate futures; extern crate libc; +#[macro_use] +extern crate downcast_rs; mod any_error; mod flags; @@ -12,6 +14,7 @@ mod libdeno; mod module_specifier; mod modules; mod ops; +mod resources; mod shared_queue; pub use crate::any_error::*; @@ -24,6 +27,7 @@ pub use crate::libdeno::PinnedBuf; pub use crate::module_specifier::*; pub use crate::modules::*; pub use crate::ops::*; +pub use crate::resources::*; pub fn v8_version() -> &'static str { use std::ffi::CStr; diff --git a/core/resources.rs b/core/resources.rs new file mode 100644 index 0000000000..c3d8b7107f --- /dev/null +++ b/core/resources.rs @@ -0,0 +1,84 @@ +// Copyright 2018-2019 the Deno authors. All rights reserved. MIT license. + +// Think of Resources as File Descriptors. They are integers that are allocated by +// the privileged side of Deno to refer to various rust objects that need to be +// referenced between multiple ops. For example, network sockets are resources. +// Resources may or may not correspond to a real operating system file +// descriptor (hence the different name). + +use downcast_rs::Downcast; +use std; +use std::any::Any; +use std::collections::HashMap; +use std::io::Error; +use std::io::ErrorKind; + +/// ResourceId is Deno's version of a file descriptor. ResourceId is also referred +/// to as rid in the code base. +pub type ResourceId = u32; + +/// These store Deno's file descriptors. These are not necessarily the operating +/// system ones. +type ResourceMap = HashMap>; + +#[derive(Default)] +pub struct ResourceTable { + map: ResourceMap, + next_id: u32, +} + +impl ResourceTable { + pub fn get(&self, rid: ResourceId) -> Result<&T, Error> { + let resource = self.map.get(&rid).ok_or_else(bad_resource)?; + let resource = &resource.downcast_ref::().ok_or_else(bad_resource)?; + Ok(resource) + } + + pub fn get_mut( + &mut self, + rid: ResourceId, + ) -> Result<&mut T, Error> { + let resource = self.map.get_mut(&rid).ok_or_else(bad_resource)?; + let resource = resource.downcast_mut::().ok_or_else(bad_resource)?; + Ok(resource) + } + + // TODO: resource id allocation should probably be randomized for security. + fn next_rid(&mut self) -> ResourceId { + let next_rid = self.next_id; + self.next_id += 1; + next_rid as ResourceId + } + + pub fn add(&mut self, resource: Box) -> ResourceId { + let rid = self.next_rid(); + let r = self.map.insert(rid, resource); + assert!(r.is_none()); + rid + } + + // close(2) is done by dropping the value. Therefore we just need to remove + // the resource from the RESOURCE_TABLE. + pub fn close(&mut self, rid: ResourceId) -> Result<(), Error> { + let repr = self.map.remove(&rid).ok_or_else(bad_resource)?; + // Give resource a chance to cleanup (notify tasks, etc.) + repr.close(); + Ok(()) + } +} + +/// Abstract type representing resource in Deno. +pub trait Resource: Downcast + Any + Send { + /// Method that allows to cleanup resource. + fn close(&self) {} + + fn inspect_repr(&self) -> &str { + unimplemented!(); + } +} +impl_downcast!(Resource); + +// TODO: probably bad error kind +pub fn bad_resource() -> Error { + Error::new(ErrorKind::NotFound, "bad resource id") +}