0
0
Fork 0
mirror of https://github.com/denoland/deno.git synced 2024-10-29 08:58:01 -04:00
denoland-deno/src/isolate.rs

543 lines
15 KiB
Rust
Raw Normal View History

// Copyright 2018 the Deno authors. All rights reserved. MIT license.
// Do not use FlatBuffers in this module.
// TODO Currently this module uses Tokio, but it would be nice if they were
// decoupled.
use deno_dir;
use errors::DenoError;
use errors::DenoResult;
use flags;
use libdeno;
use permissions::DenoPermissions;
use snapshot;
use futures::Future;
use libc::c_void;
use std;
use std::env;
use std::ffi::CStr;
use std::ffi::CString;
use std::path::Path;
use std::sync::mpsc;
use std::sync::Arc;
use std::sync::Mutex;
2018-10-02 20:47:40 -04:00
use std::time::Duration;
use std::time::Instant;
use tokio;
use tokio_util;
type DenoException<'a> = &'a str;
// Buf represents a byte array returned from a "Op".
// The message might be empty (which will be translated into a null object on
// the javascript side) or it is a heap allocated opaque sequence of bytes.
// Usually a flatbuffer message.
pub type Buf = Box<[u8]>;
// JS promises in Deno map onto a specific Future
// which yields either a DenoError or a byte array.
pub type Op = Future<Item = Buf, Error = DenoError> + Send;
// Returns (is_sync, op)
pub type Dispatch =
2018-10-02 20:47:40 -04:00
fn(isolate: &mut Isolate, buf: &[u8], data_buf: &'static mut [u8])
-> (bool, Box<Op>);
pub struct Isolate {
libdeno_isolate: *const libdeno::isolate,
dispatch: Dispatch,
rx: mpsc::Receiver<(i32, Buf)>,
2018-10-06 22:14:33 -04:00
ntasks: i32,
2018-10-02 20:47:40 -04:00
pub timeout_due: Option<Instant>,
pub state: Arc<IsolateState>,
}
// Isolate cannot be passed between threads but IsolateState can. So any state that
// needs to be accessed outside the main V8 thread should be inside IsolateState.
pub struct IsolateState {
pub dir: deno_dir::DenoDir,
pub argv: Vec<String>,
pub permissions: Mutex<DenoPermissions>,
pub flags: flags::DenoFlags,
tx: Mutex<Option<mpsc::Sender<(i32, Buf)>>>,
2018-10-05 13:21:15 -04:00
pub metrics: Mutex<Metrics>,
}
impl IsolateState {
// Thread safe.
fn send_to_js(&self, req_id: i32, buf: Buf) {
let mut g = self.tx.lock().unwrap();
let maybe_tx = g.as_mut();
assert!(maybe_tx.is_some(), "Expected tx to not be deleted.");
let tx = maybe_tx.unwrap();
tx.send((req_id, buf)).expect("tx.send error");
}
2018-10-05 13:21:15 -04:00
pub fn check_write(&self, filename: &str) -> DenoResult<()> {
let mut perm = self.permissions.lock().unwrap();
perm.check_write(filename)
}
pub fn check_env(&self) -> DenoResult<()> {
let mut perm = self.permissions.lock().unwrap();
perm.check_env()
}
pub fn check_net(&self, filename: &str) -> DenoResult<()> {
let mut perm = self.permissions.lock().unwrap();
perm.check_net(filename)
}
2018-10-05 13:21:15 -04:00
fn metrics_op_dispatched(
&self,
bytes_sent_control: u64,
bytes_sent_data: u64,
) {
let mut metrics = self.metrics.lock().unwrap();
metrics.ops_dispatched += 1;
metrics.bytes_sent_control += bytes_sent_control;
metrics.bytes_sent_data += bytes_sent_data;
}
fn metrics_op_completed(&self, bytes_received: u64) {
let mut metrics = self.metrics.lock().unwrap();
metrics.ops_completed += 1;
metrics.bytes_received += bytes_received;
}
}
#[derive(Default)]
pub struct Metrics {
pub ops_dispatched: u64,
pub ops_completed: u64,
pub bytes_sent_control: u64,
pub bytes_sent_data: u64,
pub bytes_received: u64,
}
static DENO_INIT: std::sync::Once = std::sync::ONCE_INIT;
fn empty() -> libdeno::deno_buf {
libdeno::deno_buf {
alloc_ptr: std::ptr::null_mut(),
alloc_len: 0,
data_ptr: std::ptr::null_mut(),
data_len: 0,
}
}
impl Isolate {
pub fn new(
flags: flags::DenoFlags,
argv_rest: Vec<String>,
dispatch: Dispatch,
) -> Isolate {
DENO_INIT.call_once(|| {
unsafe { libdeno::deno_init() };
});
let shared = empty(); // TODO Use shared for message passing.
let libdeno_isolate = unsafe {
libdeno::deno_new(snapshot::deno_snapshot.clone(), shared, pre_dispatch)
};
// This channel handles sending async messages back to the runtime.
let (tx, rx) = mpsc::channel::<(i32, Buf)>();
let custom_root_path;
let custom_root = match env::var("DENO_DIR") {
Ok(path) => {
custom_root_path = path;
Some(Path::new(custom_root_path.as_str()))
}
Err(_e) => None,
};
Isolate {
libdeno_isolate,
dispatch,
rx,
2018-10-06 22:14:33 -04:00
ntasks: 0,
2018-10-02 20:47:40 -04:00
timeout_due: None,
state: Arc::new(IsolateState {
dir: deno_dir::DenoDir::new(flags.reload, custom_root).unwrap(),
argv: argv_rest,
permissions: Mutex::new(DenoPermissions::new(&flags)),
flags,
tx: Mutex::new(Some(tx)),
2018-10-05 13:21:15 -04:00
metrics: Mutex::new(Metrics::default()),
}),
}
}
pub fn as_void_ptr(&mut self) -> *mut c_void {
self as *mut _ as *mut c_void
}
pub fn from_void_ptr<'a>(ptr: *mut c_void) -> &'a mut Isolate {
let ptr = ptr as *mut _;
unsafe { &mut *ptr }
}
pub fn execute(
&mut self,
js_filename: &str,
js_source: &str,
) -> Result<(), DenoException> {
let filename = CString::new(js_filename).unwrap();
let source = CString::new(js_source).unwrap();
let r = unsafe {
libdeno::deno_execute(
self.libdeno_isolate,
self.as_void_ptr(),
filename.as_ptr(),
source.as_ptr(),
)
};
if r == 0 {
let ptr = unsafe { libdeno::deno_last_exception(self.libdeno_isolate) };
let cstr = unsafe { CStr::from_ptr(ptr) };
return Err(cstr.to_str().unwrap());
}
Ok(())
}
pub fn respond(&mut self, req_id: i32, buf: Buf) {
2018-10-17 14:30:23 -04:00
self.state.metrics_op_completed(buf.len() as u64);
2018-10-18 18:28:47 -04:00
// TODO(zero-copy) Use Buf::leak(buf) to leak the heap allocated buf. And
// don't do the memcpy in ImportBuf() (in libdeno/binding.cc)
unsafe {
libdeno::deno_respond(
self.libdeno_isolate,
self.as_void_ptr(),
req_id,
buf.into(),
)
}
}
2018-10-02 20:47:40 -04:00
fn complete_op(&mut self, req_id: i32, buf: Buf) {
// Receiving a message on rx exactly corresponds to an async task
// completing.
self.ntasks_decrement();
// Call into JS with the buf.
self.respond(req_id, buf);
}
fn timeout(&mut self) {
2018-10-02 20:47:40 -04:00
let dummy_buf = libdeno::deno_buf {
alloc_ptr: 0 as *mut u8,
alloc_len: 0,
data_ptr: 0 as *mut u8,
data_len: 0,
};
unsafe {
libdeno::deno_respond(
self.libdeno_isolate,
self.as_void_ptr(),
-1,
dummy_buf,
)
}
2018-10-02 20:47:40 -04:00
}
2018-10-12 14:22:52 -04:00
fn check_promise_errors(&self) {
unsafe {
libdeno::deno_check_promise_errors(self.libdeno_isolate);
}
}
// TODO Use Park abstraction? Note at time of writing Tokio default runtime
// does not have new_with_park().
pub fn event_loop(&mut self) {
// Main thread event loop.
while !self.is_idle() {
match recv_deadline(&self.rx, self.timeout_due) {
Ok((req_id, buf)) => self.complete_op(req_id, buf),
Err(mpsc::RecvTimeoutError::Timeout) => self.timeout(),
Err(e) => panic!("recv_deadline() failed: {:?}", e),
}
2018-10-12 14:22:52 -04:00
self.check_promise_errors();
}
2018-10-12 14:22:52 -04:00
// Check on done
self.check_promise_errors();
}
fn ntasks_increment(&mut self) {
2018-10-06 22:14:33 -04:00
assert!(self.ntasks >= 0);
self.ntasks = self.ntasks + 1;
}
fn ntasks_decrement(&mut self) {
2018-10-06 22:14:33 -04:00
self.ntasks = self.ntasks - 1;
assert!(self.ntasks >= 0);
}
fn is_idle(&self) -> bool {
2018-10-06 22:14:33 -04:00
self.ntasks == 0 && self.timeout_due.is_none()
}
}
impl Drop for Isolate {
fn drop(&mut self) {
unsafe { libdeno::deno_delete(self.libdeno_isolate) }
}
}
/// Converts Rust Buf to libdeno deno_buf.
impl From<Buf> for libdeno::deno_buf {
fn from(x: Buf) -> libdeno::deno_buf {
let len = x.len();
let ptr = Box::into_raw(x);
libdeno::deno_buf {
alloc_ptr: 0 as *mut u8,
alloc_len: 0,
data_ptr: ptr as *mut u8,
data_len: len,
}
}
}
// Dereferences the C pointer into the Rust Isolate object.
extern "C" fn pre_dispatch(
user_data: *mut c_void,
req_id: i32,
control_buf: libdeno::deno_buf,
data_buf: libdeno::deno_buf,
) {
2018-10-05 13:21:15 -04:00
// for metrics
let bytes_sent_control = control_buf.data_len as u64;
let bytes_sent_data = data_buf.data_len as u64;
// control_buf is only valid for the lifetime of this call, thus is
// interpretted as a slice.
let control_slice = unsafe {
std::slice::from_raw_parts(control_buf.data_ptr, control_buf.data_len)
};
// data_buf is valid for the lifetime of the promise, thus a mutable buf with
// static lifetime.
let data_slice = unsafe {
std::slice::from_raw_parts_mut::<'static>(
data_buf.data_ptr,
data_buf.data_len,
)
};
let isolate = Isolate::from_void_ptr(user_data);
let dispatch = isolate.dispatch;
2018-10-02 20:47:40 -04:00
let (is_sync, op) = dispatch(isolate, control_slice, data_slice);
2018-10-05 13:21:15 -04:00
isolate
.state
.metrics_op_dispatched(bytes_sent_control, bytes_sent_data);
if is_sync {
// Execute op synchronously.
let buf = tokio_util::block_on(op).unwrap();
2018-10-05 13:21:15 -04:00
let buf_size = buf.len();
2018-10-17 14:30:23 -04:00
2018-10-05 13:21:15 -04:00
if buf_size != 0 {
// Set the synchronous response, the value returned from isolate.send().
isolate.respond(req_id, buf);
2018-10-17 14:30:23 -04:00
} else {
// FIXME
isolate.state.metrics_op_completed(buf.len() as u64);
}
} else {
// Execute op asynchronously.
2018-10-05 13:21:15 -04:00
let state = Arc::clone(&isolate.state);
// TODO Ideally Tokio would could tell us how many tasks are executing, but
// it cannot currently. Therefore we track top-level promises/tasks
// manually.
isolate.ntasks_increment();
let task = op
.and_then(move |buf| {
state.send_to_js(req_id, buf);
Ok(())
}).map_err(|_| ());
tokio::spawn(task);
}
}
fn recv_deadline<T>(
rx: &mpsc::Receiver<T>,
maybe_due: Option<Instant>,
) -> Result<T, mpsc::RecvTimeoutError> {
match maybe_due {
None => rx.recv().map_err(|e| e.into()),
Some(due) => {
// Subtracting two Instants causes a panic if the resulting duration
// would become negative. Avoid this.
let now = Instant::now();
let timeout = if due > now {
due - now
} else {
Duration::new(0, 0)
};
// TODO: use recv_deadline() instead of recv_timeout() when this
// feature becomes stable/available.
rx.recv_timeout(timeout)
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use futures;
#[test]
fn test_dispatch_sync() {
let argv = vec![String::from("./deno"), String::from("hello.js")];
let (flags, rest_argv, _) = flags::set_flags(argv).unwrap();
let mut isolate = Isolate::new(flags, rest_argv, dispatch_sync);
tokio_util::init(|| {
isolate
.execute(
"y.js",
r#"
const m = new Uint8Array([4, 5, 6]);
let n = libdeno.send(m);
if (!(n.byteLength === 3 &&
n[0] === 1 &&
n[1] === 2 &&
n[2] === 3)) {
throw Error("assert error");
}
"#,
2018-10-15 16:46:42 -04:00
).expect("execute error");
isolate.event_loop();
});
}
fn dispatch_sync(
2018-10-02 20:47:40 -04:00
_isolate: &mut Isolate,
control: &[u8],
data: &'static mut [u8],
) -> (bool, Box<Op>) {
assert_eq!(control[0], 4);
assert_eq!(control[1], 5);
assert_eq!(control[2], 6);
assert_eq!(data.len(), 0);
// Send back some sync response.
let vec: Vec<u8> = vec![1, 2, 3];
let control = vec.into_boxed_slice();
let op = Box::new(futures::future::ok(control));
(true, op)
}
2018-10-05 13:21:15 -04:00
#[test]
fn test_metrics_sync() {
let argv = vec![String::from("./deno"), String::from("hello.js")];
let (flags, rest_argv, _) = flags::set_flags(argv).unwrap();
let mut isolate = Isolate::new(flags, rest_argv, metrics_dispatch_sync);
2018-10-05 13:21:15 -04:00
tokio_util::init(|| {
// Verify that metrics have been properly initialized.
{
let metrics = isolate.state.metrics.lock().unwrap();
assert_eq!(metrics.ops_dispatched, 0);
assert_eq!(metrics.ops_completed, 0);
assert_eq!(metrics.bytes_sent_control, 0);
assert_eq!(metrics.bytes_sent_data, 0);
assert_eq!(metrics.bytes_received, 0);
}
isolate
.execute(
"y.js",
r#"
const control = new Uint8Array([4, 5, 6]);
const data = new Uint8Array([42, 43, 44, 45, 46]);
libdeno.send(control, data);
"#,
2018-10-15 16:46:42 -04:00
).expect("execute error");
2018-10-05 13:21:15 -04:00
isolate.event_loop();
let metrics = isolate.state.metrics.lock().unwrap();
assert_eq!(metrics.ops_dispatched, 1);
assert_eq!(metrics.ops_completed, 1);
assert_eq!(metrics.bytes_sent_control, 3);
assert_eq!(metrics.bytes_sent_data, 5);
assert_eq!(metrics.bytes_received, 4);
});
}
#[test]
fn test_metrics_async() {
let argv = vec![String::from("./deno"), String::from("hello.js")];
let (flags, rest_argv, _) = flags::set_flags(argv).unwrap();
let mut isolate = Isolate::new(flags, rest_argv, metrics_dispatch_async);
2018-10-05 13:21:15 -04:00
tokio_util::init(|| {
// Verify that metrics have been properly initialized.
{
let metrics = isolate.state.metrics.lock().unwrap();
assert_eq!(metrics.ops_dispatched, 0);
assert_eq!(metrics.ops_completed, 0);
assert_eq!(metrics.bytes_sent_control, 0);
assert_eq!(metrics.bytes_sent_data, 0);
assert_eq!(metrics.bytes_received, 0);
}
isolate
.execute(
"y.js",
r#"
const control = new Uint8Array([4, 5, 6]);
const data = new Uint8Array([42, 43, 44, 45, 46]);
let r = libdeno.send(control, data);
if (r != null) throw Error("expected null");
"#,
2018-10-15 16:46:42 -04:00
).expect("execute error");
2018-10-05 13:21:15 -04:00
// Make sure relevant metrics are updated before task is executed.
{
let metrics = isolate.state.metrics.lock().unwrap();
assert_eq!(metrics.ops_dispatched, 1);
assert_eq!(metrics.bytes_sent_control, 3);
assert_eq!(metrics.bytes_sent_data, 5);
// Note we cannot check ops_completed nor bytes_received because that
// would be a race condition. It might be nice to have use a oneshot
// with metrics_dispatch_async() to properly validate them.
}
isolate.event_loop();
// Make sure relevant metrics are updated after task is executed.
{
let metrics = isolate.state.metrics.lock().unwrap();
assert_eq!(metrics.ops_dispatched, 1);
assert_eq!(metrics.ops_completed, 1);
assert_eq!(metrics.bytes_sent_control, 3);
assert_eq!(metrics.bytes_sent_data, 5);
assert_eq!(metrics.bytes_received, 4);
}
});
}
fn metrics_dispatch_sync(
_isolate: &mut Isolate,
_control: &[u8],
_data: &'static mut [u8],
) -> (bool, Box<Op>) {
// Send back some sync response
let vec: Vec<u8> = vec![1, 2, 3, 4];
let control = vec.into_boxed_slice();
let op = Box::new(futures::future::ok(control));
(true, op)
}
fn metrics_dispatch_async(
_isolate: &mut Isolate,
_control: &[u8],
_data: &'static mut [u8],
) -> (bool, Box<Op>) {
// Send back some sync response
let vec: Vec<u8> = vec![1, 2, 3, 4];
let control = vec.into_boxed_slice();
let op = Box::new(futures::future::ok(control));
(false, op)
}
}