deno.land / x / deno@v1.28.2 / runtime / ops / worker_host.rs
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396// Copyright 2018-2022 the Deno authors. All rights reserved. MIT license.
use crate::ops::TestingFeaturesEnabled;use crate::permissions::create_child_permissions;use crate::permissions::ChildPermissionsArg;use crate::permissions::Permissions;use crate::web_worker::run_web_worker;use crate::web_worker::SendableWebWorkerHandle;use crate::web_worker::WebWorker;use crate::web_worker::WebWorkerHandle;use crate::web_worker::WebWorkerType;use crate::web_worker::WorkerControlEvent;use crate::web_worker::WorkerId;use crate::worker::FormatJsErrorFn;use deno_core::error::AnyError;use deno_core::futures::future::LocalFutureObj;use deno_core::op;
use deno_core::serde::Deserialize;use deno_core::CancelFuture;use deno_core::CancelHandle;use deno_core::Extension;use deno_core::ModuleSpecifier;use deno_core::OpState;use deno_web::JsMessageData;use log::debug;use std::cell::RefCell;use std::collections::HashMap;use std::rc::Rc;use std::sync::Arc;
pub struct CreateWebWorkerArgs { pub name: String, pub worker_id: WorkerId, pub parent_permissions: Permissions, pub permissions: Permissions, pub main_module: ModuleSpecifier, pub worker_type: WebWorkerType,}
pub type CreateWebWorkerCb = dyn Fn(CreateWebWorkerArgs) -> (WebWorker, SendableWebWorkerHandle) + Sync + Send;
pub type WorkerEventCb = dyn Fn(WebWorker) -> LocalFutureObj<'static, Result<WebWorker, AnyError>> + Sync + Send;
/// A holder for callback that is used to create a new/// WebWorker. It's a struct instead of a type alias/// because `GothamState` used in `OpState` overrides/// value if type aliases have the same underlying type#[derive(Clone)]struct CreateWebWorkerCbHolder(Arc<CreateWebWorkerCb>);
#[derive(Clone)]struct FormatJsErrorFnHolder(Option<Arc<FormatJsErrorFn>>);
#[derive(Clone)]struct PreloadModuleCbHolder(Arc<WorkerEventCb>);
#[derive(Clone)]struct PreExecuteModuleCbHolder(Arc<WorkerEventCb>);
pub struct WorkerThread { worker_handle: WebWorkerHandle, cancel_handle: Rc<CancelHandle>,
// A WorkerThread that hasn't been explicitly terminated can only be removed // from the WorkersTable once close messages have been received for both the // control and message channels. See `close_channel`. ctrl_closed: bool, message_closed: bool,}
impl WorkerThread { fn terminate(self) { // Cancel recv ops when terminating the worker, so they don't show up as // pending ops. self.cancel_handle.cancel(); }}
impl Drop for WorkerThread { fn drop(&mut self) { self.worker_handle.clone().terminate(); }}
pub type WorkersTable = HashMap<WorkerId, WorkerThread>;
pub fn init( create_web_worker_cb: Arc<CreateWebWorkerCb>, preload_module_cb: Arc<WorkerEventCb>, pre_execute_module_cb: Arc<WorkerEventCb>, format_js_error_fn: Option<Arc<FormatJsErrorFn>>,) -> Extension { Extension::builder() .state(move |state| { state.put::<WorkersTable>(WorkersTable::default()); state.put::<WorkerId>(WorkerId::default());
let create_web_worker_cb_holder = CreateWebWorkerCbHolder(create_web_worker_cb.clone()); state.put::<CreateWebWorkerCbHolder>(create_web_worker_cb_holder); let preload_module_cb_holder = PreloadModuleCbHolder(preload_module_cb.clone()); state.put::<PreloadModuleCbHolder>(preload_module_cb_holder); let pre_execute_module_cb_holder = PreExecuteModuleCbHolder(pre_execute_module_cb.clone()); state.put::<PreExecuteModuleCbHolder>(pre_execute_module_cb_holder); let format_js_error_fn_holder = FormatJsErrorFnHolder(format_js_error_fn.clone()); state.put::<FormatJsErrorFnHolder>(format_js_error_fn_holder);
Ok(()) }) .ops(vec![ op_create_worker::decl(), op_host_terminate_worker::decl(), op_host_post_message::decl(), op_host_recv_ctrl::decl(), op_host_recv_message::decl(), ]) .build()}
#[derive(Deserialize)]#[serde(rename_all = "camelCase")]pub struct CreateWorkerArgs { has_source_code: bool, name: Option<String>, permissions: Option<ChildPermissionsArg>, source_code: String, specifier: String, worker_type: WebWorkerType,}
/// Create worker as the host#[op]fn op_create_worker( state: &mut OpState, args: CreateWorkerArgs,) -> Result<WorkerId, AnyError> { let specifier = args.specifier.clone(); let maybe_source_code = if args.has_source_code { Some(args.source_code.clone()) } else { None }; let args_name = args.name; let worker_type = args.worker_type; if let WebWorkerType::Classic = worker_type { if let TestingFeaturesEnabled(false) = state.borrow() { return Err( deno_webstorage::DomExceptionNotSupportedError::new( "Classic workers are not supported.", ) .into(), ); } }
if args.permissions.is_some() { super::check_unstable(state, "Worker.deno.permissions"); } let parent_permissions = state.borrow_mut::<Permissions>(); let worker_permissions = if let Some(child_permissions_arg) = args.permissions { create_child_permissions(parent_permissions, child_permissions_arg)? } else { parent_permissions.clone() }; let parent_permissions = parent_permissions.clone(); let worker_id = state.take::<WorkerId>(); let create_web_worker_cb = state.take::<CreateWebWorkerCbHolder>(); state.put::<CreateWebWorkerCbHolder>(create_web_worker_cb.clone()); let preload_module_cb = state.take::<PreloadModuleCbHolder>(); state.put::<PreloadModuleCbHolder>(preload_module_cb.clone()); let pre_execute_module_cb = state.take::<PreExecuteModuleCbHolder>(); state.put::<PreExecuteModuleCbHolder>(pre_execute_module_cb.clone()); let format_js_error_fn = state.take::<FormatJsErrorFnHolder>(); state.put::<FormatJsErrorFnHolder>(format_js_error_fn.clone()); state.put::<WorkerId>(worker_id.next().unwrap());
let module_specifier = deno_core::resolve_url(&specifier)?; let worker_name = args_name.unwrap_or_default();
let (handle_sender, handle_receiver) = std::sync::mpsc::sync_channel::< Result<SendableWebWorkerHandle, AnyError>, >(1);
// Setup new thread let thread_builder = std::thread::Builder::new().name(format!("{}", worker_id));
// Spawn it thread_builder.spawn(move || { // Any error inside this block is terminal: // - JS worker is useless - meaning it throws an exception and can't do anything else, // all action done upon it should be noops // - newly spawned thread exits
let (worker, external_handle) = (create_web_worker_cb.0)(CreateWebWorkerArgs { name: worker_name, worker_id, parent_permissions, permissions: worker_permissions, main_module: module_specifier.clone(), worker_type, });
// Send thread safe handle from newly created worker to host thread handle_sender.send(Ok(external_handle)).unwrap(); drop(handle_sender);
// At this point the only method of communication with host // is using `worker.internal_channels`. // // Host can already push messages and interact with worker. run_web_worker( worker, module_specifier, maybe_source_code, preload_module_cb.0, pre_execute_module_cb.0, format_js_error_fn.0, ) })?;
// Receive WebWorkerHandle from newly created worker let worker_handle = handle_receiver.recv().unwrap()?;
let worker_thread = WorkerThread { worker_handle: worker_handle.into(), cancel_handle: CancelHandle::new_rc(), ctrl_closed: false, message_closed: false, };
// At this point all interactions with worker happen using thread // safe handler returned from previous function calls state .borrow_mut::<WorkersTable>() .insert(worker_id, worker_thread);
Ok(worker_id)}
#[op]fn op_host_terminate_worker(state: &mut OpState, id: WorkerId) { if let Some(worker_thread) = state.borrow_mut::<WorkersTable>().remove(&id) { worker_thread.terminate(); } else { debug!("tried to terminate non-existent worker {}", id); }}
enum WorkerChannel { Ctrl, Messages,}
/// Close a worker's channel. If this results in both of a worker's channels/// being closed, the worker will be removed from the workers table.fn close_channel( state: Rc<RefCell<OpState>>, id: WorkerId, channel: WorkerChannel,) { use std::collections::hash_map::Entry;
let mut s = state.borrow_mut(); let workers = s.borrow_mut::<WorkersTable>();
// `Worker.terminate()` might have been called already, meaning that we won't // find the worker in the table - in that case ignore. if let Entry::Occupied(mut entry) = workers.entry(id) { let terminate = { let worker_thread = entry.get_mut(); match channel { WorkerChannel::Ctrl => { worker_thread.ctrl_closed = true; worker_thread.message_closed } WorkerChannel::Messages => { worker_thread.message_closed = true; worker_thread.ctrl_closed } } };
if terminate { entry.remove().terminate(); } }}
/// Get control event from guest worker as host#[op]async fn op_host_recv_ctrl( state: Rc<RefCell<OpState>>, id: WorkerId,) -> Result<WorkerControlEvent, AnyError> { let (worker_handle, cancel_handle) = { let state = state.borrow(); let workers_table = state.borrow::<WorkersTable>(); let maybe_handle = workers_table.get(&id); if let Some(handle) = maybe_handle { (handle.worker_handle.clone(), handle.cancel_handle.clone()) } else { // If handle was not found it means worker has already shutdown return Ok(WorkerControlEvent::Close); } };
let maybe_event = worker_handle .get_control_event() .or_cancel(cancel_handle) .await; match maybe_event { Ok(Ok(Some(event))) => { // Terminal error means that worker should be removed from worker table. if let WorkerControlEvent::TerminalError(_) = &event { close_channel(state, id, WorkerChannel::Ctrl); } Ok(event) } Ok(Ok(None)) => { // If there was no event from worker it means it has already been closed. close_channel(state, id, WorkerChannel::Ctrl); Ok(WorkerControlEvent::Close) } Ok(Err(err)) => Err(err), Err(_) => { // The worker was terminated. Ok(WorkerControlEvent::Close) } }}
#[op]async fn op_host_recv_message( state: Rc<RefCell<OpState>>, id: WorkerId,) -> Result<Option<JsMessageData>, AnyError> { let (worker_handle, cancel_handle) = { let s = state.borrow(); let workers_table = s.borrow::<WorkersTable>(); let maybe_handle = workers_table.get(&id); if let Some(handle) = maybe_handle { (handle.worker_handle.clone(), handle.cancel_handle.clone()) } else { // If handle was not found it means worker has already shutdown return Ok(None); } };
let ret = worker_handle .port .recv(state.clone()) .or_cancel(cancel_handle) .await; match ret { Ok(Ok(ret)) => { if ret.is_none() { close_channel(state, id, WorkerChannel::Messages); } Ok(ret) } Ok(Err(err)) => Err(err), Err(_) => { // The worker was terminated. Ok(None) } }}
/// Post message to guest worker as host#[op]fn op_host_post_message( state: &mut OpState, id: WorkerId, data: JsMessageData,) -> Result<(), AnyError> { if let Some(worker_thread) = state.borrow::<WorkersTable>().get(&id) { debug!("post message to worker {}", id); let worker_handle = worker_thread.worker_handle.clone(); worker_handle.port.send(state, data)?; } else { debug!("tried to post message to non-existent worker {}", id); } Ok(())}
deno
Version Info
a year ago
