// Copyright 2018 the Deno authors. All rights reserved. MIT license. // Think of Resources as File Descriptors. They are integers that are allocated // by the privlaged side of Deno to refer to various resources. The simplest // example are standard file system files and stdio - but there will be other // resources added in the future that might not correspond to operating system // level File Descriptors. To avoid confusion we call them "resources" not "file // descriptors". This module implements a global resource table. Ops (AKA // handlers) look up resources by their integer id here. use errors::DenoError; use tokio_util; use tokio_write; use futures; use futures::future::Either; use futures::future::FutureResult; use futures::Poll; use std; use std::collections::HashMap; use std::io::Error; use std::io::{Read, Write}; use std::net::{Shutdown, SocketAddr}; use std::sync::atomic::AtomicIsize; use std::sync::atomic::Ordering; use std::sync::Mutex; use tokio; use tokio::io::{AsyncRead, AsyncWrite}; use tokio::net::TcpStream; use tokio_io; pub type ResourceId = i32; // Sometimes referred to RID. // These store Deno's file descriptors. These are not necessarily the operating // system ones. type ResourceTable = HashMap; lazy_static! { // Starts at 3 because stdio is [0-2]. static ref NEXT_RID: AtomicIsize = AtomicIsize::new(3); static ref RESOURCE_TABLE: Mutex = Mutex::new({ let mut m = HashMap::new(); // TODO Load these lazily during lookup? m.insert(0, Repr::Stdin(tokio::io::stdin())); m.insert(1, Repr::Stdout(tokio::io::stdout())); m.insert(2, Repr::Stderr(tokio::io::stderr())); m }); } // Internal representation of Resource. enum Repr { Stdin(tokio::io::Stdin), Stdout(tokio::io::Stdout), Stderr(tokio::io::Stderr), FsFile(tokio::fs::File), TcpListener(tokio::net::TcpListener), TcpStream(tokio::net::TcpStream), } // Abstract async file interface. // Ideally in unix, if Resource represents an OS rid, it will be the same. #[derive(Debug)] pub struct Resource { pub rid: ResourceId, } impl Resource { // TODO Should it return a Resource instead of net::TcpStream? pub fn poll_accept(&mut self) -> Poll<(TcpStream, SocketAddr), Error> { let mut table = RESOURCE_TABLE.lock().unwrap(); let maybe_repr = table.get_mut(&self.rid); match maybe_repr { None => panic!("bad rid"), Some(repr) => match repr { Repr::TcpListener(ref mut s) => s.poll_accept(), _ => panic!("Cannot accept"), }, } } // 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) { let mut table = RESOURCE_TABLE.lock().unwrap(); let r = table.remove(&self.rid); assert!(r.is_some()); } pub fn shutdown(&mut self, how: Shutdown) -> Result<(), DenoError> { let mut table = RESOURCE_TABLE.lock().unwrap(); let maybe_repr = table.get_mut(&self.rid); match maybe_repr { None => panic!("bad rid"), Some(repr) => match repr { Repr::TcpStream(ref mut f) => { TcpStream::shutdown(f, how).map_err(|err| DenoError::from(err)) } _ => panic!("Cannot shutdown"), }, } } } impl Read for Resource { fn read(&mut self, _buf: &mut [u8]) -> std::io::Result { unimplemented!(); } } impl AsyncRead for Resource { fn poll_read(&mut self, buf: &mut [u8]) -> Poll { let mut table = RESOURCE_TABLE.lock().unwrap(); let maybe_repr = table.get_mut(&self.rid); match maybe_repr { None => panic!("bad rid"), Some(repr) => match repr { Repr::FsFile(ref mut f) => f.poll_read(buf), Repr::Stdin(ref mut f) => f.poll_read(buf), Repr::TcpStream(ref mut f) => f.poll_read(buf), Repr::Stdout(_) | Repr::Stderr(_) => { panic!("Cannot read from stdout/stderr") } Repr::TcpListener(_) => panic!("Cannot read"), }, } } } impl Write for Resource { fn write(&mut self, _buf: &[u8]) -> std::io::Result { unimplemented!() } fn flush(&mut self) -> std::io::Result<()> { unimplemented!() } } impl AsyncWrite for Resource { fn poll_write(&mut self, buf: &[u8]) -> Poll { let mut table = RESOURCE_TABLE.lock().unwrap(); let maybe_repr = table.get_mut(&self.rid); match maybe_repr { None => panic!("bad rid"), Some(repr) => match repr { Repr::FsFile(ref mut f) => f.poll_write(buf), Repr::Stdout(ref mut f) => f.poll_write(buf), Repr::Stderr(ref mut f) => f.poll_write(buf), Repr::TcpStream(ref mut f) => f.poll_write(buf), Repr::Stdin(_) => panic!("Cannot write to stdin"), Repr::TcpListener(_) => panic!("Cannot write"), }, } } fn shutdown(&mut self) -> futures::Poll<(), std::io::Error> { unimplemented!() } } fn new_rid() -> ResourceId { let next_rid = NEXT_RID.fetch_add(1, Ordering::SeqCst); next_rid as ResourceId } pub fn add_fs_file(fs_file: tokio::fs::File) -> Resource { let rid = new_rid(); let mut tg = RESOURCE_TABLE.lock().unwrap(); match tg.insert(rid, Repr::FsFile(fs_file)) { Some(_) => panic!("There is already a file with that rid"), None => Resource { rid }, } } pub fn add_tcp_listener(listener: tokio::net::TcpListener) -> Resource { let rid = new_rid(); let mut tg = RESOURCE_TABLE.lock().unwrap(); let r = tg.insert(rid, Repr::TcpListener(listener)); assert!(r.is_none()); Resource { rid } } pub fn add_tcp_stream(stream: tokio::net::TcpStream) -> Resource { let rid = new_rid(); let mut tg = RESOURCE_TABLE.lock().unwrap(); let r = tg.insert(rid, Repr::TcpStream(stream)); assert!(r.is_none()); Resource { rid } } pub fn lookup(rid: ResourceId) -> Option { let table = RESOURCE_TABLE.lock().unwrap(); table.get(&rid).map(|_| Resource { rid }) } type EagerRead = Either, FutureResult<(R, T, usize), std::io::Error>>; #[cfg(windows)] #[allow(unused_mut)] pub fn eager_read(resource: Resource, mut buf: T) -> EagerRead where T: AsMut<[u8]>, { Either::A(tokio_io::io::read(resource, buf)).into() } #[cfg(not(windows))] use std::os::unix::io::{AsRawFd, FromRawFd, IntoRawFd}; #[cfg(not(windows))] fn eager_read_tcp>( tcp_stream: &TcpStream, resource: Resource, mut buf: T, ) -> EagerRead { // Unforunately we can't just call read() on tokio::net::TcpStream let fd = (*tcp_stream).as_raw_fd(); let mut std_tcp_stream = unsafe { std::net::TcpStream::from_raw_fd(fd) }; let read_result = std_tcp_stream.read(buf.as_mut()); // std_tcp_stream will close when it gets dropped. Thus... let _ = std_tcp_stream.into_raw_fd(); match read_result { Ok(nread) => Either::B(futures::future::ok((resource, buf, nread))), Err(err) => { if err.kind() == std::io::ErrorKind::WouldBlock { Either::A(tokio_io::io::read(resource, buf)) } else { Either::B(futures::future::err(err)) } } } } // This is an optimization that Tokio should do. // Attempt to call read() on the main thread. #[cfg(not(windows))] pub fn eager_read(resource: Resource, buf: T) -> EagerRead where T: AsMut<[u8]>, { let mut table = RESOURCE_TABLE.lock().unwrap(); let maybe_repr = table.get_mut(&resource.rid); match maybe_repr { None => panic!("bad rid"), Some(repr) => match repr { Repr::TcpStream(ref mut tcp_stream) => { eager_read_tcp(tcp_stream, resource, buf) } _ => Either::A(tokio_io::io::read(resource, buf)), }, } } type EagerWrite = Either, FutureResult<(R, T, usize), std::io::Error>>; #[cfg(windows)] pub fn eager_write(resource: Resource, buf: T) -> EagerWrite where T: AsRef<[u8]>, { Either::A(tokio_write::write(resource, buf)).into() } #[cfg(not(windows))] fn eager_write_tcp>( tcp_stream: &TcpStream, resource: Resource, buf: T, ) -> EagerWrite { let fd = (*tcp_stream).as_raw_fd(); let mut std_tcp_stream = unsafe { std::net::TcpStream::from_raw_fd(fd) }; let write_result = std_tcp_stream.write(buf.as_ref()); // std_tcp_stream will close when it gets dropped. Thus... let _ = std_tcp_stream.into_raw_fd(); match write_result { Ok(nwrite) => Either::B(futures::future::ok((resource, buf, nwrite))), Err(err) => { if err.kind() == std::io::ErrorKind::WouldBlock { Either::A(tokio_write::write(resource, buf)) } else { Either::B(futures::future::err(err)) } } } } // This is an optimization that Tokio should do. // Attempt to call write() on the main thread. #[cfg(not(windows))] pub fn eager_write(resource: Resource, buf: T) -> EagerWrite where T: AsRef<[u8]>, { let mut table = RESOURCE_TABLE.lock().unwrap(); let maybe_repr = table.get_mut(&resource.rid); match maybe_repr { None => panic!("bad rid"), Some(repr) => match repr { Repr::TcpStream(ref mut tcp_stream) => { eager_write_tcp(tcp_stream, resource, buf) } _ => Either::A(tokio_write::write(resource, buf)), }, } } type EagerAccept = Either< tokio_util::Accept, FutureResult<(TcpStream, SocketAddr), std::io::Error>, >; #[cfg(windows)] pub fn eager_accept(resource: Resource) -> EagerAccept { Either::A(tokio_util::accept(resource)).into() } #[cfg(not(windows))] fn eager_accept_tcp( tcp_listener: &tokio::net::TcpListener, resource: Resource, ) -> EagerAccept { let fd = (*tcp_listener).as_raw_fd(); let std_listener = unsafe { std::net::TcpListener::from_raw_fd(fd) }; let result = std_listener.accept(); // std_listener will close when it gets dropped. Thus... let _ = std_listener.into_raw_fd(); match result { Ok((std_stream, addr)) => { let result = tokio::net::TcpStream::from_std( std_stream, &tokio::reactor::Handle::default(), ); let tokio_stream = result.unwrap(); Either::B(futures::future::ok((tokio_stream, addr))) } Err(err) => { if err.kind() == std::io::ErrorKind::WouldBlock { Either::A(tokio_util::accept(resource)) } else { Either::B(futures::future::err(err)) } } } } // This is an optimization that Tokio should do. // Attempt to call write() on the main thread. #[cfg(not(windows))] pub fn eager_accept(resource: Resource) -> EagerAccept { let mut table = RESOURCE_TABLE.lock().unwrap(); let maybe_repr = table.get_mut(&resource.rid); match maybe_repr { None => panic!("bad rid"), Some(repr) => match repr { Repr::TcpListener(ref mut tcp_listener) => { eager_accept_tcp(tcp_listener, resource) } _ => Either::A(tokio_util::accept(resource)), }, } }