feat(ext/flash): An optimized http/1.1 server (#15405)

Co-authored-by: Bartek Iwańczuk <biwanczuk@gmail.com>
Co-authored-by: Ben Noordhuis <info@bnoordhuis.nl>
Co-authored-by: crowlkats <crowlkats@toaxl.com>
Co-authored-by: Ryan Dahl <ry@tinyclouds.org>

1 files changed, 1567 insertions, 0 deletions

diff --git a/ext/flash/lib.rs b/ext/flash/lib.rs
new file mode 100644
index 000000000..2c0cc548a
--- /dev/null
+++ b/ext/flash/lib.rs
@@ -0,0 +1,1567 @@
+// Copyright 2018-2022 the Deno authors. All rights reserved. MIT license.
+
+// False positive lint for explicit drops.
+// https://github.com/rust-lang/rust-clippy/issues/6446
+#![allow(clippy::await_holding_lock)]
+
+use deno_core::error::type_error;
+use deno_core::error::AnyError;
+use deno_core::op;
+use deno_core::serde_v8;
+use deno_core::v8;
+use deno_core::v8::fast_api;
+use deno_core::ByteString;
+use deno_core::CancelFuture;
+use deno_core::CancelHandle;
+use deno_core::Extension;
+use deno_core::OpState;
+use deno_core::StringOrBuffer;
+use deno_core::ZeroCopyBuf;
+use deno_core::V8_WRAPPER_OBJECT_INDEX;
+use deno_tls::load_certs;
+use deno_tls::load_private_keys;
+use http::header::HeaderName;
+use http::header::CONNECTION;
+use http::header::CONTENT_LENGTH;
+use http::header::EXPECT;
+use http::header::TRANSFER_ENCODING;
+use http::header::UPGRADE;
+use http::HeaderValue;
+use log::trace;
+use mio::net::TcpListener;
+use mio::net::TcpStream;
+use mio::Events;
+use mio::Interest;
+use mio::Poll;
+use mio::Token;
+use serde::Deserialize;
+use serde::Serialize;
+use std::cell::RefCell;
+use std::cell::UnsafeCell;
+use std::collections::HashMap;
+use std::ffi::c_void;
+use std::future::Future;
+use std::intrinsics::transmute;
+use std::io::BufReader;
+use std::io::Read;
+use std::io::Write;
+use std::marker::PhantomPinned;
+use std::mem::replace;
+use std::net::SocketAddr;
+use std::pin::Pin;
+use std::rc::Rc;
+use std::sync::Arc;
+use std::sync::Mutex;
+use std::task::Context;
+use std::time::Duration;
+use tokio::sync::mpsc;
+use tokio::task::JoinHandle;
+
+mod chunked;
+
+#[cfg(unix)]
+mod sendfile;
+
+pub struct FlashContext {
+  next_server_id: u32,
+  join_handles: HashMap<u32, JoinHandle<Result<(), AnyError>>>,
+  pub servers: HashMap<u32, ServerContext>,
+}
+
+pub struct ServerContext {
+  _addr: SocketAddr,
+  tx: mpsc::Sender<NextRequest>,
+  rx: mpsc::Receiver<NextRequest>,
+  response: HashMap<u32, NextRequest>,
+  listening_rx: Option<mpsc::Receiver<()>>,
+  close_tx: mpsc::Sender<()>,
+  cancel_handle: Rc<CancelHandle>,
+}
+
+struct InnerRequest {
+  _headers: Vec<httparse::Header<'static>>,
+  req: httparse::Request<'static, 'static>,
+  body_offset: usize,
+  body_len: usize,
+  buffer: Pin<Box<[u8]>>,
+}
+
+#[derive(Debug, PartialEq)]
+enum ParseStatus {
+  None,
+  Ongoing(usize),
+}
+
+type TlsTcpStream = rustls::StreamOwned<rustls::ServerConnection, TcpStream>;
+
+enum InnerStream {
+  Tcp(TcpStream),
+  Tls(Box<TlsTcpStream>),
+}
+
+struct Stream {
+  inner: InnerStream,
+  detached: bool,
+  read_rx: Option<mpsc::Receiver<()>>,
+  read_tx: Option<mpsc::Sender<()>>,
+  parse_done: ParseStatus,
+  buffer: UnsafeCell<Vec<u8>>,
+  read_lock: Arc<Mutex<()>>,
+  _pin: PhantomPinned,
+}
+
+impl Stream {
+  pub fn detach_ownership(&mut self) {
+    self.detached = true;
+  }
+
+  fn reattach_ownership(&mut self) {
+    self.detached = false;
+  }
+}
+
+impl Write for Stream {
+  #[inline]
+  fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
+    match self.inner {
+      InnerStream::Tcp(ref mut stream) => stream.write(buf),
+      InnerStream::Tls(ref mut stream) => stream.write(buf),
+    }
+  }
+  #[inline]
+  fn flush(&mut self) -> std::io::Result<()> {
+    match self.inner {
+      InnerStream::Tcp(ref mut stream) => stream.flush(),
+      InnerStream::Tls(ref mut stream) => stream.flush(),
+    }
+  }
+}
+
+impl Read for Stream {
+  #[inline]
+  fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
+    match self.inner {
+      InnerStream::Tcp(ref mut stream) => stream.read(buf),
+      InnerStream::Tls(ref mut stream) => stream.read(buf),
+    }
+  }
+}
+
+struct NextRequest {
+  // Pointer to stream owned by the server loop thread.
+  //
+  // Why not Arc<Mutex<Stream>>? Performance. The stream
+  // is never written to by the server loop thread.
+  //
+  // Dereferencing is safe until server thread finishes and
+  // op_flash_serve resolves or websocket upgrade is performed.
+  socket: *mut Stream,
+  inner: InnerRequest,
+  keep_alive: bool,
+  #[allow(dead_code)]
+  upgrade: bool,
+  content_read: usize,
+  content_length: Option<u64>,
+  remaining_chunk_size: Option<usize>,
+  te_chunked: bool,
+  expect_continue: bool,
+}
+
+// SAFETY: Sent from server thread to JS thread.
+// See comment above for `socket`.
+unsafe impl Send for NextRequest {}
+
+impl NextRequest {
+  #[inline(always)]
+  pub fn socket<'a>(&self) -> &'a mut Stream {
+    // SAFETY: Dereferencing is safe until server thread detaches socket or finishes.
+    unsafe { &mut *self.socket }
+  }
+}
+
+#[op]
+fn op_flash_respond(
+  op_state: &mut OpState,
+  server_id: u32,
+  token: u32,
+  response: StringOrBuffer,
+  maybe_body: Option<ZeroCopyBuf>,
+  shutdown: bool,
+) {
+  let flash_ctx = op_state.borrow_mut::<FlashContext>();
+  let ctx = flash_ctx.servers.get_mut(&server_id).unwrap();
+
+  let mut close = false;
+  let sock = match shutdown {
+    true => {
+      let tx = ctx.response.remove(&token).unwrap();
+      close = !tx.keep_alive;
+      tx.socket()
+    }
+    // In case of a websocket upgrade or streaming response.
+    false => {
+      let tx = ctx.response.get(&token).unwrap();
+      tx.socket()
+    }
+  };
+
+  sock.read_tx.take();
+  sock.read_rx.take();
+
+  let _ = sock.write(&response);
+  if let Some(response) = maybe_body {
+    let _ = sock.write(format!("{:x}", response.len()).as_bytes());
+    let _ = sock.write(b"\r\n");
+    let _ = sock.write(&response);
+    let _ = sock.write(b"\r\n");
+  }
+
+  // server is done writing and request doesn't want to kept alive.
+  if shutdown && close {
+    match &mut sock.inner {
+      InnerStream::Tcp(stream) => {
+        // Typically shutdown shouldn't fail.
+        let _ = stream.shutdown(std::net::Shutdown::Both);
+      }
+      InnerStream::Tls(stream) => {
+        let _ = stream.sock.shutdown(std::net::Shutdown::Both);
+      }
+    }
+  }
+}
+
+#[op]
+fn op_flash_respond_chuncked(
+  op_state: &mut OpState,
+  server_id: u32,
+  token: u32,
+  response: Option<ZeroCopyBuf>,
+  shutdown: bool,
+) {
+  let flash_ctx = op_state.borrow_mut::<FlashContext>();
+  let ctx = flash_ctx.servers.get_mut(&server_id).unwrap();
+  match response {
+    Some(response) => {
+      respond_chunked(ctx, token, shutdown, Some(&response));
+    }
+    None => {
+      respond_chunked(ctx, token, shutdown, None);
+    }
+  }
+}
+
+#[op]
+async fn op_flash_write_resource(
+  op_state: Rc<RefCell<OpState>>,
+  response: StringOrBuffer,
+  server_id: u32,
+  token: u32,
+  resource_id: deno_core::ResourceId,
+) -> Result<(), AnyError> {
+  let resource = op_state.borrow_mut().resource_table.take_any(resource_id)?;
+  let sock = {
+    let op_state = &mut op_state.borrow_mut();
+    let flash_ctx = op_state.borrow_mut::<FlashContext>();
+    let ctx = flash_ctx.servers.get_mut(&server_id).unwrap();
+    ctx.response.remove(&token).unwrap().socket()
+  };
+
+  drop(op_state);
+  let _ = sock.write(&response);
+
+  #[cfg(unix)]
+  {
+    use std::os::unix::io::AsRawFd;
+    if let InnerStream::Tcp(stream_handle) = &sock.inner {
+      let stream_handle = stream_handle.as_raw_fd();
+      if let Some(fd) = resource.clone().backing_fd() {
+        // SAFETY: all-zero byte-pattern is a valid value for libc::stat.
+        let mut stat: libc::stat = unsafe { std::mem::zeroed() };
+        // SAFETY: call to libc::fstat.
+        if unsafe { libc::fstat(fd, &mut stat) } >= 0 {
+          let _ = sock.write(
+            format!("Content-Length: {}\r\n\r\n", stat.st_size).as_bytes(),
+          );
+          let tx = sendfile::SendFile {
+            io: (fd, stream_handle),
+            written: 0,
+          };
+          tx.await?;
+          return Ok(());
+        }
+      }
+    }
+  }
+
+  let _ = sock.write(b"Transfer-Encoding: chunked\r\n\r\n");
+  loop {
+    let vec = vec![0u8; 64 * 1024]; // 64KB
+    let buf = ZeroCopyBuf::new_temp(vec);
+    let (nread, buf) = resource.clone().read_return(buf).await?;
+    if nread == 0 {
+      let _ = sock.write(b"0\r\n\r\n");
+      break;
+    }
+    let response = &buf[..nread];
+
+    let _ = sock.write(format!("{:x}", response.len()).as_bytes());
+    let _ = sock.write(b"\r\n");
+    let _ = sock.write(response);
+    let _ = sock.write(b"\r\n");
+  }
+
+  resource.close();
+  Ok(())
+}
+
+pub struct RespondFast;
+
+impl fast_api::FastFunction for RespondFast {
+  fn function(&self) -> *const c_void {
+    op_flash_respond_fast as *const c_void
+  }
+
+  fn args(&self) -> &'static [fast_api::Type] {
+    &[
+      fast_api::Type::V8Value,
+      fast_api::Type::Uint32,
+      fast_api::Type::TypedArray(fast_api::CType::Uint8),
+      fast_api::Type::Bool,
+    ]
+  }
+
+  fn return_type(&self) -> fast_api::CType {
+    fast_api::CType::Void
+  }
+}
+
+fn flash_respond(
+  ctx: &mut ServerContext,
+  token: u32,
+  shutdown: bool,
+  response: &[u8],
+) {
+  let mut close = false;
+  let sock = match shutdown {
+    true => {
+      let tx = ctx.response.remove(&token).unwrap();
+      close = !tx.keep_alive;
+      tx.socket()
+    }
+    // In case of a websocket upgrade or streaming response.
+    false => {
+      let tx = ctx.response.get(&token).unwrap();
+      tx.socket()
+    }
+  };
+
+  sock.read_tx.take();
+  sock.read_rx.take();
+
+  let _ = sock.write(response);
+  // server is done writing and request doesn't want to kept alive.
+  if shutdown && close {
+    match &mut sock.inner {
+      InnerStream::Tcp(stream) => {
+        // Typically shutdown shouldn't fail.
+        let _ = stream.shutdown(std::net::Shutdown::Both);
+      }
+      InnerStream::Tls(stream) => {
+        let _ = stream.sock.shutdown(std::net::Shutdown::Both);
+      }
+    }
+  }
+}
+
+unsafe fn op_flash_respond_fast(
+  recv: v8::Local<v8::Object>,
+  token: u32,
+  response: *const fast_api::FastApiTypedArray<u8>,
+  shutdown: bool,
+) {
+  let ptr =
+    recv.get_aligned_pointer_from_internal_field(V8_WRAPPER_OBJECT_INDEX);
+  let ctx = &mut *(ptr as *mut ServerContext);
+
+  let response = &*response;
+  if let Some(response) = response.get_storage_if_aligned() {
+    flash_respond(ctx, token, shutdown, response);
+  } else {
+    todo!();
+  }
+}
+
+pub struct RespondChunkedFast;
+
+impl fast_api::FastFunction for RespondChunkedFast {
+  fn function(&self) -> *const c_void {
+    op_flash_respond_chunked_fast as *const c_void
+  }
+
+  fn args(&self) -> &'static [fast_api::Type] {
+    &[
+      fast_api::Type::V8Value,
+      fast_api::Type::Uint32,
+      fast_api::Type::TypedArray(fast_api::CType::Uint8),
+      fast_api::Type::Bool,
+    ]
+  }
+
+  fn return_type(&self) -> fast_api::CType {
+    fast_api::CType::Void
+  }
+}
+
+unsafe fn op_flash_respond_chunked_fast(
+  recv: v8::Local<v8::Object>,
+  token: u32,
+  response: *const fast_api::FastApiTypedArray<u8>,
+  shutdown: bool,
+) {
+  let ptr =
+    recv.get_aligned_pointer_from_internal_field(V8_WRAPPER_OBJECT_INDEX);
+  let ctx = &mut *(ptr as *mut ServerContext);
+
+  let response = &*response;
+  if let Some(response) = response.get_storage_if_aligned() {
+    respond_chunked(ctx, token, shutdown, Some(response));
+  } else {
+    todo!();
+  }
+}
+
+fn respond_chunked(
+  ctx: &mut ServerContext,
+  token: u32,
+  shutdown: bool,
+  response: Option<&[u8]>,
+) {
+  let sock = match shutdown {
+    true => {
+      let tx = ctx.response.remove(&token).unwrap();
+      tx.socket()
+    }
+    // In case of a websocket upgrade or streaming response.
+    false => {
+      let tx = ctx.response.get(&token).unwrap();
+      tx.socket()
+    }
+  };
+
+  if let Some(response) = response {
+    let _ = sock.write(format!("{:x}", response.len()).as_bytes());
+    let _ = sock.write(b"\r\n");
+    let _ = sock.write(response);
+    let _ = sock.write(b"\r\n");
+  }
+
+  // The last chunk
+  if shutdown {
+    let _ = sock.write(b"0\r\n\r\n");
+  }
+  sock.reattach_ownership();
+}
+
+macro_rules! get_request {
+  ($op_state: ident, $token: ident) => {
+    get_request!($op_state, 0, $token)
+  };
+  ($op_state: ident, $server_id: expr, $token: ident) => {{
+    let flash_ctx = $op_state.borrow_mut::<FlashContext>();
+    let ctx = flash_ctx.servers.get_mut(&$server_id).unwrap();
+    ctx.response.get_mut(&$token).unwrap()
+  }};
+}
+
+#[repr(u32)]
+pub enum Method {
+  GET = 0,
+  HEAD,
+  CONNECT,
+  PUT,
+  DELETE,
+  OPTIONS,
+  TRACE,
+  POST,
+  PATCH,
+}
+
+#[inline]
+fn get_method(req: &mut NextRequest) -> u32 {
+  let method = match req.inner.req.method.unwrap() {
+    "GET" => Method::GET,
+    "POST" => Method::POST,
+    "PUT" => Method::PUT,
+    "DELETE" => Method::DELETE,
+    "OPTIONS" => Method::OPTIONS,
+    "HEAD" => Method::HEAD,
+    "PATCH" => Method::PATCH,
+    "TRACE" => Method::TRACE,
+    "CONNECT" => Method::CONNECT,
+    _ => Method::GET,
+  };
+  method as u32
+}
+
+#[op]
+fn op_flash_method(state: &mut OpState, server_id: u32, token: u32) -> u32 {
+  let req = get_request!(state, server_id, token);
+  get_method(req)
+}
+
+#[op]
+async fn op_flash_close_server(state: Rc<RefCell<OpState>>, server_id: u32) {
+  let close_tx = {
+    let mut op_state = state.borrow_mut();
+    let flash_ctx = op_state.borrow_mut::<FlashContext>();
+    let ctx = flash_ctx.servers.get_mut(&server_id).unwrap();
+    ctx.cancel_handle.cancel();
+    ctx.close_tx.clone()
+  };
+  let _ = close_tx.send(()).await;
+}
+
+#[op]
+fn op_flash_path(
+  state: Rc<RefCell<OpState>>,
+  server_id: u32,
+  token: u32,
+) -> String {
+  let mut op_state = state.borrow_mut();
+  let flash_ctx = op_state.borrow_mut::<FlashContext>();
+  let ctx = flash_ctx.servers.get_mut(&server_id).unwrap();
+  ctx
+    .response
+    .get(&token)
+    .unwrap()
+    .inner
+    .req
+    .path
+    .unwrap()
+    .to_string()
+}
+
+#[inline]
+fn next_request_sync(ctx: &mut ServerContext) -> u32 {
+  let mut tokens = 0;
+  while let Ok(token) = ctx.rx.try_recv() {
+    ctx.response.insert(tokens, token);
+    tokens += 1;
+  }
+  tokens
+}
+
+pub struct NextRequestFast;
+
+impl fast_api::FastFunction for NextRequestFast {
+  fn function(&self) -> *const c_void {
+    op_flash_next_fast as *const c_void
+  }
+
+  fn args(&self) -> &'static [fast_api::Type] {
+    &[fast_api::Type::V8Value]
+  }
+
+  fn return_type(&self) -> fast_api::CType {
+    fast_api::CType::Uint32
+  }
+}
+
+unsafe fn op_flash_next_fast(recv: v8::Local<v8::Object>) -> u32 {
+  let ptr =
+    recv.get_aligned_pointer_from_internal_field(V8_WRAPPER_OBJECT_INDEX);
+  let ctx = &mut *(ptr as *mut ServerContext);
+  next_request_sync(ctx)
+}
+
+pub struct GetMethodFast;
+
+impl fast_api::FastFunction for GetMethodFast {
+  fn function(&self) -> *const c_void {
+    op_flash_get_method_fast as *const c_void
+  }
+
+  fn args(&self) -> &'static [fast_api::Type] {
+    &[fast_api::Type::V8Value, fast_api::Type::Uint32]
+  }
+
+  fn return_type(&self) -> fast_api::CType {
+    fast_api::CType::Uint32
+  }
+}
+
+unsafe fn op_flash_get_method_fast(
+  recv: v8::Local<v8::Object>,
+  token: u32,
+) -> u32 {
+  let ptr =
+    recv.get_aligned_pointer_from_internal_field(V8_WRAPPER_OBJECT_INDEX);
+  let ctx = &mut *(ptr as *mut ServerContext);
+  let req = ctx.response.get_mut(&token).unwrap();
+  get_method(req)
+}
+
+// Fast calls
+#[op(v8)]
+fn op_flash_make_request<'scope>(
+  scope: &mut v8::HandleScope<'scope>,
+  state: &mut OpState,
+) -> serde_v8::Value<'scope> {
+  let object_template = v8::ObjectTemplate::new(scope);
+  assert!(object_template
+    .set_internal_field_count((V8_WRAPPER_OBJECT_INDEX + 1) as usize));
+  let obj = object_template.new_instance(scope).unwrap();
+  let ctx = {
+    let flash_ctx = state.borrow_mut::<FlashContext>();
+    let ctx = flash_ctx.servers.get_mut(&0).unwrap();
+    ctx as *mut ServerContext
+  };
+  obj.set_aligned_pointer_in_internal_field(V8_WRAPPER_OBJECT_INDEX, ctx as _);
+
+  // nextRequest
+  {
+    let builder = v8::FunctionTemplate::builder(
+      |_: &mut v8::HandleScope,
+       args: v8::FunctionCallbackArguments,
+       mut rv: v8::ReturnValue| {
+        let external: v8::Local<v8::External> =
+          args.data().unwrap().try_into().unwrap();
+        // SAFETY: This external is guaranteed to be a pointer to a ServerContext
+        let ctx = unsafe { &mut *(external.value() as *mut ServerContext) };
+        rv.set_uint32(next_request_sync(ctx));
+      },
+    )
+    .data(v8::External::new(scope, ctx as *mut _).into());
+
+    let func = builder.build_fast(scope, &NextRequestFast, None);
+    let func: v8::Local<v8::Value> = func.get_function(scope).unwrap().into();
+
+    let key = v8::String::new(scope, "nextRequest").unwrap();
+    obj.set(scope, key.into(), func).unwrap();
+  }
+
+  // getMethod
+  {
+    let builder = v8::FunctionTemplate::builder(
+      |scope: &mut v8::HandleScope,
+       args: v8::FunctionCallbackArguments,
+       mut rv: v8::ReturnValue| {
+        let external: v8::Local<v8::External> =
+          args.data().unwrap().try_into().unwrap();
+        // SAFETY: This external is guaranteed to be a pointer to a ServerContext
+        let ctx = unsafe { &mut *(external.value() as *mut ServerContext) };
+        let token = args.get(0).uint32_value(scope).unwrap();
+        let req = ctx.response.get_mut(&token).unwrap();
+        rv.set_uint32(get_method(req));
+      },
+    )
+    .data(v8::External::new(scope, ctx as *mut _).into());
+
+    let func = builder.build_fast(scope, &GetMethodFast, None);
+    let func: v8::Local<v8::Value> = func.get_function(scope).unwrap().into();
+
+    let key = v8::String::new(scope, "getMethod").unwrap();
+    obj.set(scope, key.into(), func).unwrap();
+  }
+
+  // respondChunked
+  {
+    let builder = v8::FunctionTemplate::builder(
+      |scope: &mut v8::HandleScope,
+       args: v8::FunctionCallbackArguments,
+       _: v8::ReturnValue| {
+        let external: v8::Local<v8::External> =
+          args.data().unwrap().try_into().unwrap();
+        // SAFETY: This external is guaranteed to be a pointer to a ServerContext
+        let ctx = unsafe { &mut *(external.value() as *mut ServerContext) };
+
+        let token = args.get(0).uint32_value(scope).unwrap();
+
+        let response: v8::Local<v8::ArrayBufferView> =
+          args.get(1).try_into().unwrap();
+        let ab = response.buffer(scope).unwrap();
+        let store = ab.get_backing_store();
+        let (offset, len) = (response.byte_offset(), response.byte_length());
+        // SAFETY: v8::SharedRef<v8::BackingStore> is similar to Arc<[u8]>,
+        // it points to a fixed continuous slice of bytes on the heap.
+        // We assume it's initialized and thus safe to read (though may not contain meaningful data)
+        let response = unsafe {
+          &*(&store[offset..offset + len] as *const _ as *const [u8])
+        };
+
+        let shutdown = args.get(2).boolean_value(scope);
+
+        respond_chunked(ctx, token, shutdown, Some(response));
+      },
+    )
+    .data(v8::External::new(scope, ctx as *mut _).into());
+
+    let func = builder.build_fast(scope, &RespondChunkedFast, None);
+    let func: v8::Local<v8::Value> = func.get_function(scope).unwrap().into();
+
+    let key = v8::String::new(scope, "respondChunked").unwrap();
+    obj.set(scope, key.into(), func).unwrap();
+  }
+
+  // respond
+  {
+    let builder = v8::FunctionTemplate::builder(
+      |scope: &mut v8::HandleScope,
+       args: v8::FunctionCallbackArguments,
+       _: v8::ReturnValue| {
+        let external: v8::Local<v8::External> =
+          args.data().unwrap().try_into().unwrap();
+        // SAFETY: This external is guaranteed to be a pointer to a ServerContext
+        let ctx = unsafe { &mut *(external.value() as *mut ServerContext) };
+
+        let token = args.get(0).uint32_value(scope).unwrap();
+
+        let response: v8::Local<v8::ArrayBufferView> =
+          args.get(1).try_into().unwrap();
+        let ab = response.buffer(scope).unwrap();
+        let store = ab.get_backing_store();
+        let (offset, len) = (response.byte_offset(), response.byte_length());
+        // SAFETY: v8::SharedRef<v8::BackingStore> is similar to Arc<[u8]>,
+        // it points to a fixed continuous slice of bytes on the heap.
+        // We assume it's initialized and thus safe to read (though may not contain meaningful data)
+        let response = unsafe {
+          &*(&store[offset..offset + len] as *const _ as *const [u8])
+        };
+
+        let shutdown = args.get(2).boolean_value(scope);
+
+        flash_respond(ctx, token, shutdown, response);
+      },
+    )
+    .data(v8::External::new(scope, ctx as *mut _).into());
+
+    let func = builder.build_fast(scope, &RespondFast, None);
+    let func: v8::Local<v8::Value> = func.get_function(scope).unwrap().into();
+
+    let key = v8::String::new(scope, "respond").unwrap();
+    obj.set(scope, key.into(), func).unwrap();
+  }
+
+  let value: v8::Local<v8::Value> = obj.into();
+  value.into()
+}
+
+#[inline]
+fn has_body_stream(req: &NextRequest) -> bool {
+  let sock = req.socket();
+  sock.read_rx.is_some()
+}
+
+#[op]
+fn op_flash_has_body_stream(
+  op_state: &mut OpState,
+  server_id: u32,
+  token: u32,
+) -> bool {
+  let req = get_request!(op_state, server_id, token);
+  has_body_stream(req)
+}
+
+#[op]
+fn op_flash_headers(
+  state: Rc<RefCell<OpState>>,
+  server_id: u32,
+  token: u32,
+) -> Result<Vec<(ByteString, ByteString)>, AnyError> {
+  let mut op_state = state.borrow_mut();
+  let flash_ctx = op_state.borrow_mut::<FlashContext>();
+  let ctx = flash_ctx
+    .servers
+    .get_mut(&server_id)
+    .ok_or_else(|| type_error("server closed"))?;
+  let inner_req = &ctx
+    .response
+    .get(&token)
+    .ok_or_else(|| type_error("request closed"))?
+    .inner
+    .req;
+  Ok(
+    inner_req
+      .headers
+      .iter()
+      .map(|h| (h.name.as_bytes().into(), h.value.into()))
+      .collect(),
+  )
+}
+
+// Remember the first packet we read? It probably also has some body data. This op quickly copies it into
+// a buffer and sets up channels for streaming the rest.
+#[op]
+fn op_flash_first_packet(
+  op_state: &mut OpState,
+  server_id: u32,
+  token: u32,
+) -> Result<Option<ZeroCopyBuf>, AnyError> {
+  let tx = get_request!(op_state, server_id, token);
+  let sock = tx.socket();
+
+  if !tx.te_chunked && tx.content_length.is_none() {
+    return Ok(None);
+  }
+
+  if tx.expect_continue {
+    let _ = sock.write(b"HTTP/1.1 100 Continue\r\n\r\n");
+    tx.expect_continue = false;
+  }
+
+  let buffer = &tx.inner.buffer[tx.inner.body_offset..tx.inner.body_len];
+  // Oh there is nothing here.
+  if buffer.is_empty() {
+    return Ok(Some(ZeroCopyBuf::empty()));
+  }
+
+  if tx.te_chunked {
+    let mut buf = vec![0; 1024];
+    let mut offset = 0;
+    let mut decoder = chunked::Decoder::new(
+      std::io::Cursor::new(buffer),
+      tx.remaining_chunk_size,
+    );
+
+    loop {
+      match decoder.read(&mut buf[offset..]) {
+        Ok(n) => {
+          tx.remaining_chunk_size = decoder.remaining_chunks_size;
+          offset += n;
+
+          if n == 0 {
+            tx.te_chunked = false;
+            buf.truncate(offset);
+            return Ok(Some(buf.into()));
+          }
+
+          if offset < buf.len()
+            && decoder.source.position() < buffer.len() as u64
+          {
+            continue;
+          }
+
+          buf.truncate(offset);
+          return Ok(Some(buf.into()));
+        }
+        Err(e) => {
+          return Err(type_error(format!("{}", e)));
+        }
+      }
+    }
+  }
+
+  tx.content_length
+    .ok_or_else(|| type_error("no content-length"))?;
+  tx.content_read += buffer.len();
+
+  Ok(Some(buffer.to_vec().into()))
+}
+
+#[op]
+async fn op_flash_read_body(
+  state: Rc<RefCell<OpState>>,
+  server_id: u32,
+  token: u32,
+  mut buf: ZeroCopyBuf,
+) -> usize {
+  // SAFETY: we cannot hold op_state borrow across the await point. The JS caller
+  // is responsible for ensuring this is not called concurrently.
+  let ctx = unsafe {
+    {
+      let op_state = &mut state.borrow_mut();
+      let flash_ctx = op_state.borrow_mut::<FlashContext>();
+      flash_ctx.servers.get_mut(&server_id).unwrap() as *mut ServerContext
+    }
+    .as_mut()
+    .unwrap()
+  };
+  let tx = ctx.response.get_mut(&token).unwrap();
+
+  if tx.te_chunked {
+    let mut decoder =
+      chunked::Decoder::new(tx.socket(), tx.remaining_chunk_size);
+    loop {
+      let sock = tx.socket();
+
+      let _lock = sock.read_lock.lock().unwrap();
+      match decoder.read(&mut buf) {
+        Ok(n) => {
+          tx.remaining_chunk_size = decoder.remaining_chunks_size;
+          return n;
+        }
+        Err(e) if e.kind() == std::io::ErrorKind::InvalidInput => {
+          panic!("chunked read error: {}", e);
+        }
+        Err(_) => {
+          drop(_lock);
+          sock.read_rx.as_mut().unwrap().recv().await.unwrap();
+        }
+      }
+    }
+  }
+
+  if let Some(content_length) = tx.content_length {
+    let sock = tx.socket();
+    let l = sock.read_lock.clone();
+
+    loop {
+      let _lock = l.lock().unwrap();
+      if tx.content_read >= content_length as usize {
+        return 0;
+      }
+      match sock.read(&mut buf) {
+        Ok(n) => {
+          tx.content_read += n;
+          return n;
+        }
+        _ => {
+          drop(_lock);
+          sock.read_rx.as_mut().unwrap().recv().await.unwrap();
+        }
+      }
+    }
+  }
+
+  0
+}
+
+// https://github.com/hyperium/hyper/blob/0c8ee93d7f557afc63ca2a5686d19071813ab2b7/src/headers.rs#L67
+#[inline]
+fn from_digits(bytes: &[u8]) -> Option<u64> {
+  // cannot use FromStr for u64, since it allows a signed prefix
+  let mut result = 0u64;
+  const RADIX: u64 = 10;
+  if bytes.is_empty() {
+    return None;
+  }
+  for &b in bytes {
+    // can't use char::to_digit, since we haven't verified these bytes
+    // are utf-8.
+    match b {
+      b'0'..=b'9' => {
+        result = result.checked_mul(RADIX)?;
+        result = result.checked_add((b - b'0') as u64)?;
+      }
+      _ => {
+        return None;
+      }
+    }
+  }
+  Some(result)
+}
+
+#[inline]
+fn connection_has(value: &HeaderValue, needle: &str) -> bool {
+  if let Ok(s) = value.to_str() {
+    for val in s.split(',') {
+      if val.trim().eq_ignore_ascii_case(needle) {
+        return true;
+      }
+    }
+  }
+  false
+}
+
+#[derive(Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ListenOpts {
+  cert: Option<String>,
+  key: Option<String>,
+  hostname: String,
+  port: u16,
+  use_tls: bool,
+}
+
+fn run_server(
+  tx: mpsc::Sender<NextRequest>,
+  listening_tx: mpsc::Sender<()>,
+  mut close_rx: mpsc::Receiver<()>,
+  addr: SocketAddr,
+  maybe_cert: Option<String>,
+  maybe_key: Option<String>,
+) -> Result<(), AnyError> {
+  let mut listener = TcpListener::bind(addr)?;
+  let mut poll = Poll::new()?;
+  let token = Token(0);
+  poll
+    .registry()
+    .register(&mut listener, token, Interest::READABLE)
+    .unwrap();
+
+  let tls_context: Option<Arc<rustls::ServerConfig>> = {
+    if let Some(cert) = maybe_cert {
+      let key = maybe_key.unwrap();
+      let certificate_chain: Vec<rustls::Certificate> =
+        load_certs(&mut BufReader::new(cert.as_bytes()))?;
+      let private_key = load_private_keys(key.as_bytes())?.remove(0);
+
+      let config = rustls::ServerConfig::builder()
+        .with_safe_defaults()
+        .with_no_client_auth()
+        .with_single_cert(certificate_chain, private_key)
+        .expect("invalid key or certificate");
+      Some(Arc::new(config))
+    } else {
+      None
+    }
+  };
+
+  listening_tx.blocking_send(()).unwrap();
+  let mut sockets = HashMap::with_capacity(1000);
+  let mut counter: usize = 1;
+  let mut events = Events::with_capacity(1024);
+  'outer: loop {
+    let result = close_rx.try_recv();
+    if result.is_ok() {
+      break 'outer;
+    }
+    // FIXME(bartlomieju): how does Tokio handle it? I just put random 100ms
+    // timeout here to handle close signal.
+    match poll.poll(&mut events, Some(Duration::from_millis(100))) {
+      Err(ref e) if e.kind() == std::io::ErrorKind::Interrupted => continue,
+      Err(e) => panic!("{}", e),
+      Ok(()) => (),
+    }
+    'events: for event in &events {
+      if close_rx.try_recv().is_ok() {
+        break 'outer;
+      }
+      let token = event.token();
+      match token {
+        Token(0) => loop {
+          match listener.accept() {
+            Ok((mut socket, _)) => {
+              counter += 1;
+              let token = Token(counter);
+              poll
+                .registry()
+                .register(&mut socket, token, Interest::READABLE)
+                .unwrap();
+              let socket = match tls_context {
+                Some(ref tls_conf) => {
+                  let connection =
+                    rustls::ServerConnection::new(tls_conf.clone()).unwrap();
+                  InnerStream::Tls(Box::new(rustls::StreamOwned::new(
+                    connection, socket,
+                  )))
+                }
+                None => InnerStream::Tcp(socket),
+              };
+              let stream = Box::pin(Stream {
+                inner: socket,
+                detached: false,
+                read_rx: None,
+                read_tx: None,
+                read_lock: Arc::new(Mutex::new(())),
+                parse_done: ParseStatus::None,
+                buffer: UnsafeCell::new(vec![0_u8; 1024]),
+                _pin: PhantomPinned,
+              });
+
+              trace!("New connection: {}", token.0);
+              sockets.insert(token, stream);
+            }
+            Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => break,
+            Err(_) => break,
+          }
+        },
+        token => {
+          let socket = sockets.get_mut(&token).unwrap();
+          // SAFETY: guarantee that we will never move the data out of the mutable reference.
+          let socket = unsafe {
+            let mut_ref: Pin<&mut Stream> = Pin::as_mut(socket);
+            Pin::get_unchecked_mut(mut_ref)
+          };
+          let sock_ptr = socket as *mut _;
+
+          if socket.detached {
+            match &mut socket.inner {
+              InnerStream::Tcp(ref mut socket) => {
+                poll.registry().deregister(socket).unwrap();
+              }
+              InnerStream::Tls(_) => {
+                todo!("upgrade tls not implemented");
+              }
+            }
+
+            let boxed = sockets.remove(&token).unwrap();
+            std::mem::forget(boxed);
+            trace!("Socket detached: {}", token.0);
+            continue;
+          }
+
+          debug_assert!(event.is_readable());
+
+          trace!("Socket readable: {}", token.0);
+          if let Some(tx) = &socket.read_tx {
+            {
+              let _l = socket.read_lock.lock().unwrap();
+            }
+            trace!("Sending readiness notification: {}", token.0);
+            let _ = tx.blocking_send(());
+
+            continue;
+          }
+
+          let mut headers = vec![httparse::EMPTY_HEADER; 40];
+          let mut req = httparse::Request::new(&mut headers);
+          let body_offset;
+          let body_len;
+          loop {
+            // SAFETY: It is safe for the read buf to be mutable here.
+            let buffer = unsafe { &mut *socket.buffer.get() };
+            let offset = match socket.parse_done {
+              ParseStatus::None => 0,
+              ParseStatus::Ongoing(offset) => offset,
+            };
+            if offset >= buffer.len() {
+              buffer.resize(offset * 2, 0);
+            }
+            let nread = socket.read(&mut buffer[offset..]);
+
+            match nread {
+              Ok(0) => {
+                trace!("Socket closed: {}", token.0);
+                // FIXME: don't remove while JS is writing!
+                // sockets.remove(&token);
+                continue 'events;
+              }
+              Ok(read) => match req.parse(&buffer[..offset + read]) {
+                Ok(httparse::Status::Complete(n)) => {
+                  body_offset = n;
+                  body_len = offset + read;
+                  socket.parse_done = ParseStatus::None;
+                  break;
+                }
+                Ok(httparse::Status::Partial) => {
+                  socket.parse_done = ParseStatus::Ongoing(offset + read);
+                  continue;
+                }
+                Err(_) => {
+                  let _ = socket.write(b"HTTP/1.1 400 Bad Request\r\n\r\n");
+                  continue 'events;
+                }
+              },
+              Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => {
+                break 'events
+              }
+              Err(_) => break 'events,
+            }
+          }
+
+          debug_assert_eq!(socket.parse_done, ParseStatus::None);
+          if let Some(method) = &req.method {
+            if method == &"POST" || method == &"PUT" {
+              let (tx, rx) = mpsc::channel(100);
+              socket.read_tx = Some(tx);
+              socket.read_rx = Some(rx);
+            }
+          }
+
+          // SAFETY: It is safe for the read buf to be mutable here.
+          let buffer = unsafe { &mut *socket.buffer.get() };
+          let inner_req = InnerRequest {
+            // SAFETY: backing buffer is pinned and lives as long as the request.
+            req: unsafe { transmute::<httparse::Request<'_, '_>, _>(req) },
+            // SAFETY: backing buffer is pinned and lives as long as the request.
+            _headers: unsafe {
+              transmute::<Vec<httparse::Header<'_>>, _>(headers)
+            },
+            buffer: Pin::new(
+              replace(buffer, vec![0_u8; 1024]).into_boxed_slice(),
+            ),
+            body_offset,
+            body_len,
+          };
+          // h1
+          // https://github.com/tiny-http/tiny-http/blob/master/src/client.rs#L177
+          // https://github.com/hyperium/hyper/blob/4545c3ef191ce9b5f5d250ee27c4c96f9b71d2c6/src/proto/h1/role.rs#L127
+          let mut keep_alive = inner_req.req.version.unwrap() == 1;
+          let mut upgrade = false;
+          let mut expect_continue = false;
+          let mut te = false;
+          let mut te_chunked = false;
+          let mut content_length = None;
+          for header in inner_req.req.headers.iter() {
+            match HeaderName::from_bytes(header.name.as_bytes()) {
+              Ok(CONNECTION) => {
+                // SAFETY: illegal bytes are validated by httparse.
+                let value = unsafe {
+                  HeaderValue::from_maybe_shared_unchecked(header.value)
+                };
+                if keep_alive {
+                  // 1.1
+                  keep_alive = !connection_has(&value, "close");
+                } else {
+                  // 1.0
+                  keep_alive = connection_has(&value, "keep-alive");
+                }
+              }
+              Ok(UPGRADE) => {
+                upgrade = inner_req.req.version.unwrap() == 1;
+              }
+              Ok(TRANSFER_ENCODING) => {
+                // https://tools.ietf.org/html/rfc7230#section-3.3.3
+                debug_assert!(inner_req.req.version.unwrap() == 1);
+                // Two states for Transfer-Encoding because we want to make sure Content-Length handling knows it.
+                te = true;
+                content_length = None;
+                // SAFETY: illegal bytes are validated by httparse.
+                let value = unsafe {
+                  HeaderValue::from_maybe_shared_unchecked(header.value)
+                };
+                if let Ok(Some(encoding)) =
+                  value.to_str().map(|s| s.rsplit(',').next())
+                {
+                  // Chunked must always be the last encoding
+                  if encoding.trim().eq_ignore_ascii_case("chunked") {
+                    te_chunked = true;
+                  }
+                }
+              }
+              // Transfer-Encoding overrides the Content-Length.
+              Ok(CONTENT_LENGTH) if !te => {
+                if let Some(len) = from_digits(header.value) {
+                  if let Some(prev) = content_length {
+                    if prev != len {
+                      let _ = socket.write(b"HTTP/1.1 400 Bad Request\r\n\r\n");
+                      continue 'events;
+                    }
+                    continue;
+                  }
+                  content_length = Some(len);
+                } else {
+                  let _ = socket.write(b"HTTP/1.1 400 Bad Request\r\n\r\n");
+                  continue 'events;
+                }
+              }
+              Ok(EXPECT) if inner_req.req.version.unwrap() != 0 => {
+                expect_continue =
+                  header.value.eq_ignore_ascii_case(b"100-continue");
+              }
+              _ => {}
+            }
+          }
+
+          // There is Transfer-Encoding but its not chunked.
+          if te && !te_chunked {
+            let _ = socket.write(b"HTTP/1.1 400 Bad Request\r\n\r\n");
+            continue 'events;
+          }
+
+          tx.blocking_send(NextRequest {
+            socket: sock_ptr,
+            // SAFETY: headers backing buffer outlives the mio event loop ('static)
+            inner: inner_req,
+            keep_alive,
+            upgrade,
+            te_chunked,
+            remaining_chunk_size: None,
+            content_read: 0,
+            content_length,
+            expect_continue,
+          })
+          .ok();
+        }
+      }
+    }
+  }
+
+  Ok(())
+}
+
+#[op]
+fn op_flash_serve<P>(
+  state: &mut OpState,
+  opts: ListenOpts,
+) -> Result<u32, AnyError>
+where
+  P: FlashPermissions + 'static,
+{
+  if opts.use_tls {
+    check_unstable(state, "Deno.serveTls");
+  } else {
+    check_unstable(state, "Deno.serve");
+  }
+  state
+    .borrow_mut::<P>()
+    .check_net(&(&opts.hostname, Some(opts.port)))?;
+  let addr = SocketAddr::new(opts.hostname.parse()?, opts.port);
+  let (tx, rx) = mpsc::channel(100);
+  let (close_tx, close_rx) = mpsc::channel(1);
+  let (listening_tx, listening_rx) = mpsc::channel(1);
+  let ctx = ServerContext {
+    _addr: addr,
+    tx,
+    rx,
+    response: HashMap::with_capacity(1000),
+    close_tx,
+    listening_rx: Some(listening_rx),
+    cancel_handle: CancelHandle::new_rc(),
+  };
+  let tx = ctx.tx.clone();
+  let maybe_cert = opts.cert;
+  let maybe_key = opts.key;
+  let join_handle = tokio::task::spawn_blocking(move || {
+    run_server(tx, listening_tx, close_rx, addr, maybe_cert, maybe_key)
+  });
+  let flash_ctx = state.borrow_mut::<FlashContext>();
+  let server_id = flash_ctx.next_server_id;
+  flash_ctx.next_server_id += 1;
+  flash_ctx.join_handles.insert(server_id, join_handle);
+  flash_ctx.servers.insert(server_id, ctx);
+  Ok(server_id)
+}
+
+#[op]
+fn op_flash_wait_for_listening(
+  state: &mut OpState,
+  server_id: u32,
+) -> Result<impl Future<Output = Result<(), AnyError>> + 'static, AnyError> {
+  let mut listening_rx = {
+    let flash_ctx = state.borrow_mut::<FlashContext>();
+    let server_ctx = flash_ctx
+      .servers
+      .get_mut(&server_id)
+      .ok_or_else(|| type_error("server not found"))?;
+    server_ctx.listening_rx.take().unwrap()
+  };
+  Ok(async move {
+    listening_rx.recv().await;
+    Ok(())
+  })
+}
+
+#[op]
+fn op_flash_drive_server(
+  state: &mut OpState,
+  server_id: u32,
+) -> Result<impl Future<Output = Result<(), AnyError>> + 'static, AnyError> {
+  let join_handle = {
+    let flash_ctx = state.borrow_mut::<FlashContext>();
+    flash_ctx
+      .join_handles
+      .remove(&server_id)
+      .ok_or_else(|| type_error("server not found"))?
+  };
+  Ok(async move {
+    join_handle
+      .await
+      .map_err(|_| type_error("server join error"))??;
+    Ok(())
+  })
+}
+
+// Asychronous version of op_flash_next. This can be a bottleneck under
+// heavy load, it should be used as a fallback if there are no buffered
+// requests i.e `op_flash_next() == 0`.
+#[op]
+async fn op_flash_next_async(
+  op_state: Rc<RefCell<OpState>>,
+  server_id: u32,
+) -> u32 {
+  let ctx = {
+    let mut op_state = op_state.borrow_mut();
+    let flash_ctx = op_state.borrow_mut::<FlashContext>();
+    let ctx = flash_ctx.servers.get_mut(&server_id).unwrap();
+    ctx as *mut ServerContext
+  };
+  // SAFETY: we cannot hold op_state borrow across the await point. The JS caller
+  // is responsible for ensuring this is not called concurrently.
+  let ctx = unsafe { &mut *ctx };
+  let cancel_handle = &ctx.cancel_handle;
+  let mut tokens = 0;
+  while let Ok(token) = ctx.rx.try_recv() {
+    ctx.response.insert(tokens, token);
+    tokens += 1;
+  }
+  if tokens == 0 {
+    if let Ok(Some(req)) = ctx.rx.recv().or_cancel(cancel_handle).await {
+      ctx.response.insert(tokens, req);
+      tokens += 1;
+    }
+  }
+  tokens
+}
+
+// Syncrhonous version of op_flash_next_async. Under heavy load,
+// this can collect buffered requests from rx channel and return tokens in a single batch.
+//
+// perf: please do not add any arguments to this op. With optimizations enabled,
+// the ContextScope creation is optimized away and the op is as simple as:
+//   f(info: *const v8::FunctionCallbackInfo) { let rv = ...; rv.set_uint32(op_flash_next()); }
+#[op]
+fn op_flash_next(op_state: &mut OpState) -> u32 {
+  let flash_ctx = op_state.borrow_mut::<FlashContext>();
+  let ctx = flash_ctx.servers.get_mut(&0).unwrap();
+  next_request_sync(ctx)
+}
+
+// Syncrhonous version of op_flash_next_async. Under heavy load,
+// this can collect buffered requests from rx channel and return tokens in a single batch.
+#[op]
+fn op_flash_next_server(op_state: &mut OpState, server_id: u32) -> u32 {
+  let flash_ctx = op_state.borrow_mut::<FlashContext>();
+  let ctx = flash_ctx.servers.get_mut(&server_id).unwrap();
+  next_request_sync(ctx)
+}
+
+// Wrapper type for tokio::net::TcpStream that implements
+// deno_websocket::UpgradedStream
+struct UpgradedStream(tokio::net::TcpStream);
+impl tokio::io::AsyncRead for UpgradedStream {
+  fn poll_read(
+    self: Pin<&mut Self>,
+    cx: &mut Context,
+    buf: &mut tokio::io::ReadBuf,
+  ) -> std::task::Poll<std::result::Result<(), std::io::Error>> {
+    Pin::new(&mut self.get_mut().0).poll_read(cx, buf)
+  }
+}
+
+impl tokio::io::AsyncWrite for UpgradedStream {
+  fn poll_write(
+    self: Pin<&mut Self>,
+    cx: &mut Context,
+    buf: &[u8],
+  ) -> std::task::Poll<Result<usize, std::io::Error>> {
+    Pin::new(&mut self.get_mut().0).poll_write(cx, buf)
+  }
+  fn poll_flush(
+    self: Pin<&mut Self>,
+    cx: &mut Context,
+  ) -> std::task::Poll<Result<(), std::io::Error>> {
+    Pin::new(&mut self.get_mut().0).poll_flush(cx)
+  }
+  fn poll_shutdown(
+    self: Pin<&mut Self>,
+    cx: &mut Context,
+  ) -> std::task::Poll<Result<(), std::io::Error>> {
+    Pin::new(&mut self.get_mut().0).poll_shutdown(cx)
+  }
+}
+
+impl deno_websocket::Upgraded for UpgradedStream {}
+
+#[inline]
+pub fn detach_socket(
+  ctx: &mut ServerContext,
+  token: u32,
+) -> Result<tokio::net::TcpStream, AnyError> {
+  // Two main 'hacks' to get this working:
+  //   * make server thread forget about the socket. `detach_ownership` prevents the socket from being
+  //      dropped on the server thread.
+  //   * conversion from mio::net::TcpStream -> tokio::net::TcpStream.  There is no public API so we
+  //      use raw fds.
+  let tx = ctx
+    .response
+    .remove(&token)
+    .ok_or_else(|| type_error("request closed"))?;
+  let stream = tx.socket();
+  // prevent socket from being dropped on server thread.
+  // TODO(@littledivy): Box-ify, since there is no overhead.
+  stream.detach_ownership();
+
+  #[cfg(unix)]
+  let std_stream = {
+    use std::os::unix::prelude::AsRawFd;
+    use std::os::unix::prelude::FromRawFd;
+    let fd = match stream.inner {
+      InnerStream::Tcp(ref tcp) => tcp.as_raw_fd(),
+      _ => todo!(),
+    };
+    // SAFETY: `fd` is a valid file descriptor.
+    unsafe { std::net::TcpStream::from_raw_fd(fd) }
+  };
+  #[cfg(windows)]
+  let std_stream = {
+    use std::os::windows::prelude::AsRawSocket;
+    use std::os::windows::prelude::FromRawSocket;
+    let fd = match stream.inner {
+      InnerStream::Tcp(ref tcp) => tcp.as_raw_socket(),
+      _ => todo!(),
+    };
+    // SAFETY: `fd` is a valid file descriptor.
+    unsafe { std::net::TcpStream::from_raw_socket(fd) }
+  };
+  let stream = tokio::net::TcpStream::from_std(std_stream)?;
+  Ok(stream)
+}
+
+#[op]
+async fn op_flash_upgrade_websocket(
+  state: Rc<RefCell<OpState>>,
+  server_id: u32,
+  token: u32,
+) -> Result<deno_core::ResourceId, AnyError> {
+  let stream = {
+    let op_state = &mut state.borrow_mut();
+    let flash_ctx = op_state.borrow_mut::<FlashContext>();
+    detach_socket(flash_ctx.servers.get_mut(&server_id).unwrap(), token)?
+  };
+  deno_websocket::ws_create_server_stream(
+    &state,
+    Box::pin(UpgradedStream(stream)),
+  )
+  .await
+}
+
+pub struct Unstable(pub bool);
+
+fn check_unstable(state: &OpState, api_name: &str) {
+  let unstable = state.borrow::<Unstable>();
+
+  if !unstable.0 {
+    eprintln!(
+      "Unstable API '{}'. The --unstable flag must be provided.",
+      api_name
+    );
+    std::process::exit(70);
+  }
+}
+
+pub trait FlashPermissions {
+  fn check_net<T: AsRef<str>>(
+    &mut self,
+    _host: &(T, Option<u16>),
+  ) -> Result<(), AnyError>;
+}
+
+pub fn init<P: FlashPermissions + 'static>(unstable: bool) -> Extension {
+  Extension::builder()
+    .js(deno_core::include_js_files!(
+      prefix "deno:ext/flash",
+      "01_http.js",
+    ))
+    .ops(vec![
+      op_flash_serve::decl::<P>(),
+      op_flash_respond::decl(),
+      op_flash_respond_chuncked::decl(),
+      op_flash_method::decl(),
+      op_flash_path::decl(),
+      op_flash_headers::decl(),
+      op_flash_next::decl(),
+      op_flash_next_server::decl(),
+      op_flash_next_async::decl(),
+      op_flash_read_body::decl(),
+      op_flash_upgrade_websocket::decl(),
+      op_flash_drive_server::decl(),
+      op_flash_wait_for_listening::decl(),
+      op_flash_first_packet::decl(),
+      op_flash_has_body_stream::decl(),
+      op_flash_close_server::decl(),
+      op_flash_make_request::decl(),
+      op_flash_write_resource::decl(),
+    ])
+    .state(move |op_state| {
+      op_state.put(Unstable(unstable));
+      op_state.put(FlashContext {
+        next_server_id: 0,
+        join_handles: HashMap::default(),
+        servers: HashMap::default(),
+      });
+      Ok(())
+    })
+    .build()
+}