1
0
Fork 0
mirror of https://github.com/denoland/deno.git synced 2024-12-22 07:14:47 -05:00
denoland-deno/cli/ops/worker_host.rs
Ryan Dahl ed680552a2
fix: basic web worker message passing (#3893)
Removes OP_HOST_GET_WORKER_LOADED, OP_HOST_POLL_WORKER, 
OP_HOST_RESUME_WORKER and ready/messageBuffer in cli/js/workers.ts.
2020-02-05 17:16:07 -05:00

234 lines
6.6 KiB
Rust

// Copyright 2018-2020 the Deno authors. All rights reserved. MIT license.
use super::dispatch_json::{Deserialize, JsonOp, Value};
use crate::deno_error::bad_resource;
use crate::deno_error::js_check;
use crate::deno_error::DenoError;
use crate::deno_error::ErrorKind;
use crate::ops::dispatch_json::JsonResult;
use crate::ops::json_op;
use crate::startup_data;
use crate::state::ThreadSafeState;
use crate::web_worker::WebWorker;
use deno_core::*;
use futures;
use futures::future::FutureExt;
use futures::future::TryFutureExt;
use std;
use std::convert::From;
use std::sync::atomic::Ordering;
pub fn init(i: &mut Isolate, s: &ThreadSafeState) {
i.register_op(
"create_worker",
s.core_op(json_op(s.stateful_op(op_create_worker))),
);
i.register_op(
"host_close_worker",
s.core_op(json_op(s.stateful_op(op_host_close_worker))),
);
i.register_op(
"host_post_message",
s.core_op(json_op(s.stateful_op(op_host_post_message))),
);
i.register_op(
"host_get_message",
s.core_op(json_op(s.stateful_op(op_host_get_message))),
);
i.register_op("metrics", s.core_op(json_op(s.stateful_op(op_metrics))));
}
#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
struct CreateWorkerArgs {
name: Option<String>,
specifier: String,
has_source_code: bool,
source_code: String,
}
/// Create worker as the host
fn op_create_worker(
state: &ThreadSafeState,
args: Value,
_data: Option<ZeroCopyBuf>,
) -> Result<JsonOp, ErrBox> {
let args: CreateWorkerArgs = serde_json::from_value(args)?;
let specifier = args.specifier.clone();
let has_source_code = args.has_source_code;
let source_code = args.source_code.clone();
let args_name = args.name;
let parent_state = state.clone();
let (load_sender, load_receiver) =
std::sync::mpsc::sync_channel::<JsonResult>(1);
std::thread::spawn(move || {
// TODO(bartlomieju): Isn't this wrong?
let result = ModuleSpecifier::resolve_url_or_path(&specifier);
if let Err(err) = result {
load_sender.send(Err(err.into())).unwrap();
return;
}
let module_specifier = if !has_source_code {
let referrer = parent_state.main_module.to_string();
let result = ModuleSpecifier::resolve_import(&specifier, &referrer);
if let Err(err) = result {
load_sender.send(Err(err.into())).unwrap();
return;
}
result.unwrap()
} else {
result.unwrap()
};
let result = ThreadSafeState::new_for_worker(
parent_state.global_state.clone(),
Some(parent_state.permissions.clone()), // by default share with parent
module_specifier.clone(),
);
if let Err(err) = result {
load_sender.send(Err(err)).unwrap();
return;
}
let child_state = result.unwrap();
let worker_name = args_name.unwrap_or_else(|| {
// TODO(bartlomieju): change it to something more descriptive
format!("USER-WORKER-{}", specifier)
});
// TODO: add a new option to make child worker not sharing permissions
// with parent (aka .clone(), requests from child won't reflect in parent)
let mut worker = WebWorker::new(
worker_name.to_string(),
startup_data::deno_isolate_init(),
child_state,
);
let script = format!("bootstrapWorkerRuntime(\"{}\")", worker_name);
js_check(worker.execute(&script));
js_check(worker.execute("runWorkerMessageLoop()"));
let worker_id = parent_state.add_child_worker(&worker);
// Has provided source code, execute immediately.
if has_source_code {
js_check(worker.execute(&source_code));
load_sender.send(Ok(json!({ "id": worker_id }))).unwrap();
return;
}
let fut = async move {
let r = worker
.execute_mod_async(&module_specifier, None, false)
.await;
if r.is_ok() {
let _ = (&mut *worker).await;
}
}
.boxed_local();
load_sender.send(Ok(json!({ "id": worker_id }))).unwrap();
crate::tokio_util::run_basic(fut);
});
let r = load_receiver.recv().unwrap();
Ok(JsonOp::Sync(r.unwrap()))
}
#[derive(Deserialize)]
struct WorkerArgs {
id: i32,
}
fn op_host_close_worker(
state: &ThreadSafeState,
args: Value,
_data: Option<ZeroCopyBuf>,
) -> Result<JsonOp, ErrBox> {
let args: WorkerArgs = serde_json::from_value(args)?;
let id = args.id as u32;
let state_ = state.clone();
let mut workers_table = state_.workers.lock().unwrap();
let maybe_worker_handle = workers_table.remove(&id);
if let Some(worker_handle) = maybe_worker_handle {
let mut sender = worker_handle.sender.clone();
sender.close_channel();
let mut receiver =
futures::executor::block_on(worker_handle.receiver.lock());
receiver.close();
};
Ok(JsonOp::Sync(json!({})))
}
#[derive(Deserialize)]
struct HostGetMessageArgs {
id: i32,
}
/// Get message from guest worker as host
fn op_host_get_message(
state: &ThreadSafeState,
args: Value,
_data: Option<ZeroCopyBuf>,
) -> Result<JsonOp, ErrBox> {
let args: HostGetMessageArgs = serde_json::from_value(args)?;
let state_ = state.clone();
let id = args.id as u32;
let mut table = state_.workers.lock().unwrap();
// TODO: don't return bad resource anymore
let worker_handle = table.get_mut(&id).ok_or_else(bad_resource)?;
let fut = worker_handle.get_message();
let op = async move {
let maybe_buf = fut.await;
Ok(json!({ "data": maybe_buf }))
};
Ok(JsonOp::Async(op.boxed_local()))
}
#[derive(Deserialize)]
struct HostPostMessageArgs {
id: i32,
}
/// Post message to guest worker as host
fn op_host_post_message(
state: &ThreadSafeState,
args: Value,
data: Option<ZeroCopyBuf>,
) -> Result<JsonOp, ErrBox> {
let args: HostPostMessageArgs = serde_json::from_value(args)?;
let id = args.id as u32;
let msg = Vec::from(data.unwrap().as_ref()).into_boxed_slice();
debug!("post message to worker {}", id);
let mut table = state.workers.lock().unwrap();
// TODO: don't return bad resource anymore
let worker_handle = table.get_mut(&id).ok_or_else(bad_resource)?;
let fut = worker_handle
.post_message(msg)
.map_err(|e| DenoError::new(ErrorKind::Other, e.to_string()));
futures::executor::block_on(fut)?;
Ok(JsonOp::Sync(json!({})))
}
fn op_metrics(
state: &ThreadSafeState,
_args: Value,
_zero_copy: Option<ZeroCopyBuf>,
) -> Result<JsonOp, ErrBox> {
let m = &state.metrics;
Ok(JsonOp::Sync(json!({
"opsDispatched": m.ops_dispatched.load(Ordering::SeqCst) as u64,
"opsCompleted": m.ops_completed.load(Ordering::SeqCst) as u64,
"bytesSentControl": m.bytes_sent_control.load(Ordering::SeqCst) as u64,
"bytesSentData": m.bytes_sent_data.load(Ordering::SeqCst) as u64,
"bytesReceived": m.bytes_received.load(Ordering::SeqCst) as u64
})))
}