#[macro_use]
extern crate derive_deref;
#[macro_use]
extern crate log;
use deno_core::Isolate as CoreIsolate;
use deno_core::*;
use futures::future::poll_fn;
use futures::prelude::*;
use futures::task::Context;
use futures::task::Poll;
use std::cell::RefCell;
use std::convert::TryInto;
use std::env;
use std::fmt::Debug;
use std::io::Error;
use std::io::ErrorKind;
use std::mem::size_of;
use std::net::SocketAddr;
use std::pin::Pin;
use std::ptr;
use std::rc::Rc;
use tokio::io::AsyncRead;
use tokio::io::AsyncWrite;
use tokio::net::TcpListener;
use tokio::net::TcpStream;
struct Logger;
impl log::Log for Logger {
fn enabled(&self, metadata: &log::Metadata) -> bool {
metadata.level() <= log::max_level()
}
fn log(&self, record: &log::Record) {
if self.enabled(record.metadata()) {
println!("{} - {}", record.level(), record.args());
}
}
fn flush(&self) {}
}
#[derive(Copy, Clone, Debug, PartialEq)]
struct Record {
pub promise_id: u32,
pub rid: u32,
pub result: i32,
}
type RecordBuf = [u8; size_of::<Record>()];
impl From<&[u8]> for Record {
fn from(buf: &[u8]) -> Self {
assert_eq!(buf.len(), size_of::<RecordBuf>());
unsafe { *(buf as *const _ as *const RecordBuf) }.into()
}
}
impl From<RecordBuf> for Record {
fn from(buf: RecordBuf) -> Self {
unsafe {
#[allow(clippy::cast_ptr_alignment)]
ptr::read_unaligned(&buf as *const _ as *const Self)
}
}
}
impl From<Record> for RecordBuf {
fn from(record: Record) -> Self {
unsafe { ptr::read(&record as *const _ as *const Self) }
}
}
struct Isolate {
core_isolate: Box<CoreIsolate>, state: State,
}
#[derive(Clone, Default, Deref)]
struct State(Rc<RefCell<StateInner>>);
#[derive(Default)]
struct StateInner {
resource_table: ResourceTable,
}
impl Isolate {
pub fn new() -> Self {
let startup_data = StartupData::Script(Script {
source: include_str!("http_bench.js"),
filename: "http_bench.js",
});
let mut isolate = Self {
core_isolate: CoreIsolate::new(startup_data, false),
state: Default::default(),
};
isolate.register_op("listen", op_listen);
isolate.register_op("accept", op_accept);
isolate.register_op("read", op_read);
isolate.register_op("write", op_write);
isolate.register_op("close", op_close);
isolate
}
fn register_op<F>(
&mut self,
name: &'static str,
handler: impl Fn(State, u32, Option<ZeroCopyBuf>) -> F + Copy + 'static,
) where
F: TryFuture,
F::Ok: TryInto<i32>,
<F::Ok as TryInto<i32>>::Error: Debug,
{
let state = self.state.clone();
let core_handler =
move |control_buf: &[u8], zero_copy_buf: Option<ZeroCopyBuf>| -> CoreOp {
let state = state.clone();
let record = Record::from(control_buf);
let is_sync = record.promise_id == 0;
let fut = async move {
let op = handler(state, record.rid, zero_copy_buf);
let result = op
.map_ok(|r| r.try_into().expect("op result does not fit in i32"))
.unwrap_or_else(|_| -1)
.await;
Ok(RecordBuf::from(Record { result, ..record })[..].into())
};
if is_sync {
Op::Sync(futures::executor::block_on(fut).unwrap())
} else {
Op::Async(fut.boxed_local())
}
};
self.core_isolate.register_op(name, core_handler);
}
}
impl Future for Isolate {
type Output = <CoreIsolate as Future>::Output;
fn poll(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> {
self.core_isolate.poll_unpin(cx)
}
}
fn op_close(
state: State,
rid: u32,
_buf: Option<ZeroCopyBuf>,
) -> impl TryFuture<Ok = u32, Error = Error> {
debug!("close rid={}", rid);
async move {
let resource_table = &mut state.borrow_mut().resource_table;
resource_table
.close(rid)
.map(|_| 0)
.ok_or_else(bad_resource)
}
}
fn op_listen(
state: State,
_rid: u32,
_buf: Option<ZeroCopyBuf>,
) -> impl TryFuture<Ok = u32, Error = Error> {
debug!("listen");
async move {
let addr = "127.0.0.1:4544".parse::<SocketAddr>().unwrap();
let listener = tokio::net::TcpListener::bind(&addr).await?;
let resource_table = &mut state.borrow_mut().resource_table;
let rid = resource_table.add("tcpListener", Box::new(listener));
Ok(rid)
}
}
fn op_accept(
state: State,
rid: u32,
_buf: Option<ZeroCopyBuf>,
) -> impl TryFuture<Ok = u32, Error = Error> {
debug!("accept rid={}", rid);
poll_fn(move |cx| {
let resource_table = &mut state.borrow_mut().resource_table;
let listener = resource_table
.get_mut::<TcpListener>(rid)
.ok_or_else(bad_resource)?;
listener.poll_accept(cx).map_ok(|(stream, _addr)| {
resource_table.add("tcpStream", Box::new(stream))
})
})
}
fn op_read(
state: State,
rid: u32,
buf: Option<ZeroCopyBuf>,
) -> impl TryFuture<Ok = usize, Error = Error> {
let mut buf = buf.unwrap();
debug!("read rid={}", rid);
poll_fn(move |cx| {
let resource_table = &mut state.borrow_mut().resource_table;
let stream = resource_table
.get_mut::<TcpStream>(rid)
.ok_or_else(bad_resource)?;
Pin::new(stream).poll_read(cx, &mut buf)
})
}
fn op_write(
state: State,
rid: u32,
buf: Option<ZeroCopyBuf>,
) -> impl TryFuture<Ok = usize, Error = Error> {
let buf = buf.unwrap();
debug!("write rid={}", rid);
poll_fn(move |cx| {
let resource_table = &mut state.borrow_mut().resource_table;
let stream = resource_table
.get_mut::<TcpStream>(rid)
.ok_or_else(bad_resource)?;
Pin::new(stream).poll_write(cx, &buf)
})
}
fn bad_resource() -> Error {
Error::new(ErrorKind::NotFound, "bad resource id")
}
fn main() {
log::set_logger(&Logger).unwrap();
log::set_max_level(
env::args()
.find(|a| a == "-D")
.map(|_| log::LevelFilter::Debug)
.unwrap_or(log::LevelFilter::Warn),
);
deno_core::v8_set_flags(env::args().collect());
let isolate = Isolate::new();
let mut runtime = tokio::runtime::Builder::new()
.basic_scheduler()
.enable_all()
.build()
.unwrap();
runtime.block_on(isolate).expect("unexpected isolate error");
}
#[test]
fn test_record_from() {
let expected = Record {
promise_id: 1,
rid: 3,
result: 4,
};
let buf = RecordBuf::from(expected);
if cfg!(target_endian = "little") {
assert_eq!(buf, [1u8, 0, 0, 0, 3, 0, 0, 0, 4, 0, 0, 0]);
}
let actual = Record::from(buf);
assert_eq!(actual, expected);
}