Skip to content

Commit

Permalink
fix: basic web worker message passing (#3893)
Browse files Browse the repository at this point in the history 
Removes OP_HOST_GET_WORKER_LOADED, OP_HOST_POLL_WORKER, 
OP_HOST_RESUME_WORKER and ready/messageBuffer in cli/js/workers.ts.
  • Loading branch information
@ry
ry authored Feb 5, 2020
1 parent 700f5e4 commit ed68055
Show file tree
Hide file tree
Showing 7 changed files with 56 additions and 167 deletions.
3 changes: 0 additions & 3 deletions cli/js/dispatch.ts
View file
Original file line number Diff line number Diff line change
Expand Up @@ -42,11 +42,8 @@ export let OP_QUERY_PERMISSION: number;
export let OP_REVOKE_PERMISSION: number;
export let OP_REQUEST_PERMISSION: number;
export let OP_CREATE_WORKER: number;
export let OP_HOST_GET_WORKER_LOADED: number;
export let OP_HOST_POST_MESSAGE: number;
export let OP_HOST_POLL_WORKER: number;
export let OP_HOST_CLOSE_WORKER: number;
export let OP_HOST_RESUME_WORKER: number;
export let OP_HOST_GET_MESSAGE: number;
export let OP_WORKER_POST_MESSAGE: number;
export let OP_WORKER_GET_MESSAGE: number;
Expand Down
71 changes: 13 additions & 58 deletions cli/js/workers.ts
View file
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,7 @@ function createWorker(
hasSourceCode: boolean,
sourceCode: Uint8Array,
name?: string
): { id: number; loaded: boolean } {
): { id: number } {
return sendSync(dispatch.OP_CREATE_WORKER, {
specifier,
hasSourceCode,
Expand All @@ -38,22 +38,6 @@ function createWorker(
});
}

async function hostGetWorkerLoaded(id: number): Promise<any> {
return await sendAsync(dispatch.OP_HOST_GET_WORKER_LOADED, { id });
}

async function hostPollWorker(id: number): Promise<any> {
return await sendAsync(dispatch.OP_HOST_POLL_WORKER, { id });
}

function hostCloseWorker(id: number): void {
sendSync(dispatch.OP_HOST_CLOSE_WORKER, { id });
}

function hostResumeWorker(id: number): void {
sendSync(dispatch.OP_HOST_RESUME_WORKER, { id });
}

function hostPostMessage(id: number, data: any): void {
const dataIntArray = encodeMessage(data);
sendSync(dispatch.OP_HOST_POST_MESSAGE, { id }, dataIntArray);
Expand Down Expand Up @@ -85,8 +69,6 @@ export interface WorkerOptions {
export class WorkerImpl extends EventTarget implements Worker {
private readonly id: number;
private isClosing = false;
private messageBuffer: any[] = [];
private ready = false;
public onerror?: (e: any) => void;
public onmessage?: (data: any) => void;
public onmessageerror?: () => void;
Expand Down Expand Up @@ -125,14 +107,13 @@ export class WorkerImpl extends EventTarget implements Worker {
}
*/

const { id, loaded } = createWorker(
const { id } = createWorker(
specifier,
hasSourceCode,
sourceCode,
options?.name
);
this.id = id;
this.ready = loaded;
this.poll();
}

Expand All @@ -158,27 +139,19 @@ export class WorkerImpl extends EventTarget implements Worker {
}

async poll(): Promise<void> {
// If worker has not been immediately executed
// then let's await it's readiness
if (!this.ready) {
const result = await hostGetWorkerLoaded(this.id);

if (result.error) {
if (!this.handleError(result.error)) {
throw new Error(result.error.message);
}
return;
while (!this.isClosing) {
const data = await hostGetMessage(this.id);
if (data == null) {
log("worker got null message. quitting.");
break;
}
if (this.onmessage) {
const event = { data };
this.onmessage(event);
}
}

// drain messages
for (const data of this.messageBuffer) {
hostPostMessage(this.id, data);
}
this.messageBuffer = [];
this.ready = true;
this.run();

/*
while (true) {
const result = await hostPollWorker(this.id);
Expand All @@ -194,32 +167,14 @@ export class WorkerImpl extends EventTarget implements Worker {
break;
}
}
*/
}

postMessage(data: any): void {
if (!this.ready) {
this.messageBuffer.push(data);
return;
}

hostPostMessage(this.id, data);
}

terminate(): void {
throw new Error("Not yet implemented");
}

private async run(): Promise<void> {
while (!this.isClosing) {
const data = await hostGetMessage(this.id);
if (data == null) {
log("worker got null message. quitting.");
break;
}
if (this.onmessage) {
const event = { data };
this.onmessage(event);
}
}
}
}
113 changes: 7 additions & 106 deletions cli/ops/worker_host.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -4,19 +4,15 @@ use crate::deno_error::bad_resource;
use crate::deno_error::js_check;
use crate::deno_error::DenoError;
use crate::deno_error::ErrorKind;
use crate::fmt_errors::JSError;
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::channel::mpsc;
use futures::future::FutureExt;
use futures::future::TryFutureExt;
use futures::sink::SinkExt;
use futures::stream::StreamExt;
use std;
use std::convert::From;
use std::sync::atomic::Ordering;
Expand All @@ -26,22 +22,10 @@ pub fn init(i: &mut Isolate, s: &ThreadSafeState) {
"create_worker",
s.core_op(json_op(s.stateful_op(op_create_worker))),
);
i.register_op(
"host_get_worker_loaded",
s.core_op(json_op(s.stateful_op(op_host_get_worker_loaded))),
);
i.register_op(
"host_poll_worker",
s.core_op(json_op(s.stateful_op(op_host_poll_worker))),
);
i.register_op(
"host_close_worker",
s.core_op(json_op(s.stateful_op(op_host_close_worker))),
);
i.register_op(
"host_resume_worker",
s.core_op(json_op(s.stateful_op(op_host_resume_worker))),
);
i.register_op(
"host_post_message",
s.core_op(json_op(s.stateful_op(op_host_post_message))),
Expand Down Expand Up @@ -130,29 +114,21 @@ fn op_create_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, "loaded": true})))
.unwrap();
load_sender.send(Ok(json!({ "id": worker_id }))).unwrap();
return;
}

let (mut sender, receiver) = mpsc::channel::<Result<(), ErrBox>>(1);

// TODO(bartlomieju): this future should be spawned on the separate thread,
// dedicated to that worker
let fut = async move {
let result = worker
let r = worker
.execute_mod_async(&module_specifier, None, false)
.await;
sender.send(result).await.expect("Failed to send message");
if r.is_ok() {
let _ = (&mut *worker).await;
}
}
.boxed_local();
let mut table = parent_state.loading_workers.lock().unwrap();
table.insert(worker_id, receiver);

load_sender
.send(Ok(json!({"id": worker_id, "loaded": false})))
.unwrap();
load_sender.send(Ok(json!({ "id": worker_id }))).unwrap();

crate::tokio_util::run_basic(fut);
});
Expand All @@ -162,67 +138,11 @@ fn op_create_worker(
Ok(JsonOp::Sync(r.unwrap()))
}

fn serialize_worker_result(result: Result<(), ErrBox>) -> Value {
use crate::deno_error::GetErrorKind;

if let Err(error) = result {
match error.kind() {
ErrorKind::JSError => {
let error = error.downcast::<JSError>().unwrap();
let exception: V8Exception = error.into();
json!({"error": {
"message": exception.message,
"fileName": exception.script_resource_name,
"lineNumber": exception.line_number,
"columnNumber": exception.start_column,
}})
}
_ => json!({"error": {
"message": error.to_string(),
}}),
}
} else {
json!({"ok": true})
}
}

#[derive(Deserialize)]
struct WorkerArgs {
id: i32,
}

fn op_host_get_worker_loaded(
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 mut table = state.loading_workers.lock().unwrap();
let mut receiver = table.remove(&id).unwrap();

let op = async move {
let result = receiver.next().await.unwrap();
Ok(serialize_worker_result(result))
};

Ok(JsonOp::Async(op.boxed_local()))
}

fn op_host_poll_worker(
_state: &ThreadSafeState,
_args: Value,
_data: Option<ZeroCopyBuf>,
) -> Result<JsonOp, ErrBox> {
println!("op_host_poll_worker");
// TOOO(ry) remove this.
todo!()
/*
let op = async { Ok(serialize_worker_result(Ok(()))) };
Ok(JsonOp::Async(op.boxed_local()))
*/
}

fn op_host_close_worker(
state: &ThreadSafeState,
args: Value,
Expand All @@ -246,25 +166,6 @@ fn op_host_close_worker(
Ok(JsonOp::Sync(json!({})))
}

fn op_host_resume_worker(
_state: &ThreadSafeState,
_args: Value,
_data: Option<ZeroCopyBuf>,
) -> Result<JsonOp, ErrBox> {
// TODO(ry) We are not on the same thread. We cannot just call worker.execute.
// We can only send messages. This needs to be reimplemented somehow.
todo!()
/*
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 worker = workers_table.get_mut(&id).unwrap();
js_check(worker.execute("runWorkerMessageLoop()"));
Ok(JsonOp::Sync(json!({})))
*/
}

#[derive(Deserialize)]
struct HostGetMessageArgs {
id: i32,
Expand All @@ -284,7 +185,7 @@ fn op_host_get_message(
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.unwrap();
let maybe_buf = fut.await;
Ok(json!({ "data": maybe_buf }))
};
Ok(JsonOp::Async(op.boxed_local()))
Expand Down
5 changes: 5 additions & 0 deletions cli/tests/integration_tests.rs
View file
Original file line number Diff line number Diff line change
Expand Up @@ -401,6 +401,11 @@ itest!(_026_workers {
});
*/

itest!(workers_basic {
args: "run --reload workers_basic.ts",
output: "workers_basic.out",
});

itest!(_027_redirect_typescript {
args: "run --reload 027_redirect_typescript.ts",
output: "027_redirect_typescript.ts.out",
Expand Down
17 changes: 17 additions & 0 deletions cli/tests/subdir/test_worker_basic.js
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,17 @@
console.log("hello from test_worker_basic.js");

// TODO(bartlomieju): add test for throwing in web worker
if (self.name !== "jsWorker") {
throw Error(`Bad worker name: ${self.name}, expected jsWorker`);
}

onmessage = function(e) {
console.log("jsWorker onmessage", e.data);
postMessage(e.data);
close();
};

onerror = function() {
console.log("called onerror in worker");
return false;
};
3 changes: 3 additions & 0 deletions cli/tests/workers_basic.out
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,3 @@
hello from test_worker_basic.js
jsWorker onmessage msg1
main recv: msg1
11 changes: 11 additions & 0 deletions cli/tests/workers_basic.ts
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,11 @@
// Tests basic postMessage, close, onmessage
const jsWorker = new Worker("./subdir/test_worker_basic.js", {
type: "module",
name: "jsWorker"
});

jsWorker.onmessage = (e): void => {
console.log("main recv: " + e.data);
};

jsWorker.postMessage("msg1");

0 comments on commit ed68055

Please sign in to comment.