1
0
Fork 0
mirror of https://github.com/denoland/deno.git synced 2024-12-24 08:09:08 -05:00
denoland-deno/tests/unit_node/worker_threads_test.ts
Marvin Hagemeister e8d731c05f
fix(node/worker_threads): data url not encoded properly with eval (#27184)
When using the `eval` option on Node's `worker_threads` the code is
passed as a `data:` URL. But we didn't encode the actual code for that,
which lead to syntax errors when including characters not allowed in an
URL.

Fixes a part of https://github.com/denoland/deno/issues/27167
2024-12-05 14:30:43 +01:00

843 lines
22 KiB
TypeScript

// Copyright 2018-2024 the Deno authors. All rights reserved. MIT license.
import {
assert,
assertEquals,
assertObjectMatch,
assertThrows,
fail,
} from "@std/assert";
import { fromFileUrl, relative, SEPARATOR } from "@std/path";
import * as workerThreads from "node:worker_threads";
import { EventEmitter, once } from "node:events";
import process from "node:process";
Deno.test("[node/worker_threads] BroadcastChannel is exported", () => {
assertEquals<unknown>(workerThreads.BroadcastChannel, BroadcastChannel);
});
Deno.test("[node/worker_threads] MessageChannel are MessagePort are exported", () => {
assert(workerThreads.MessageChannel);
assertEquals<unknown>(workerThreads.MessagePort, MessagePort);
});
Deno.test({
name: "[node/worker_threads] isMainThread",
fn() {
assertEquals(workerThreads.isMainThread, true);
},
});
Deno.test({
name: "[node/worker_threads] threadId",
fn() {
assertEquals(workerThreads.threadId, 0);
},
});
Deno.test({
name: "[node/worker_threads] resourceLimits",
fn() {
assertObjectMatch(workerThreads.resourceLimits, {});
},
});
Deno.test({
name: "[node/worker_threads] parentPort",
fn() {
assertEquals(workerThreads.parentPort, null);
},
});
Deno.test({
name: "[node/worker_threads] workerData",
fn() {
assertEquals(workerThreads.workerData, null);
},
});
Deno.test({
name: "[node/worker_threads] setEnvironmentData / getEnvironmentData",
fn() {
workerThreads.setEnvironmentData("test", "test");
assertEquals(workerThreads.getEnvironmentData("test"), "test");
},
});
Deno.test({
name: "[node/worker_threads] Worker threadId",
async fn() {
const worker = new workerThreads.Worker(
new URL("./testdata/worker_threads.mjs", import.meta.url),
);
worker.postMessage("Hello, how are you my thread?");
await once(worker, "message");
const message = await once(worker, "message");
assertEquals(message[0].threadId, 1);
worker.terminate();
const worker1 = new workerThreads.Worker(
new URL("./testdata/worker_threads.mjs", import.meta.url),
);
worker1.postMessage("Hello, how are you my thread?");
await once(worker1, "message");
assertEquals((await once(worker1, "message"))[0].threadId, 2);
worker1.terminate();
},
});
Deno.test({
name: "[node/worker_threads] Worker basics",
async fn() {
workerThreads.setEnvironmentData("test", "test");
workerThreads.setEnvironmentData(1, {
test: "random",
random: "test",
});
const { port1 } = new MessageChannel();
const worker = new workerThreads.Worker(
new URL("./testdata/worker_threads.mjs", import.meta.url),
{
workerData: ["hey", true, false, 2, port1],
// deno-lint-ignore no-explicit-any
transferList: [port1 as any],
},
);
worker.postMessage("Hello, how are you my thread?");
assertEquals((await once(worker, "message"))[0], "I'm fine!");
const data = (await once(worker, "message"))[0];
// data.threadId can be 1 when this test is run individually
if (data.threadId === 1) data.threadId = 3;
assertObjectMatch(data, {
isMainThread: false,
threadId: 3,
workerData: ["hey", true, false, 2],
envData: ["test", { test: "random", random: "test" }],
});
worker.terminate();
},
sanitizeResources: false,
});
Deno.test({
name: "[node/worker_threads] Worker eval",
async fn() {
const worker = new workerThreads.Worker(
`
import { parentPort } from "node:worker_threads";
parentPort.postMessage("It works!");
`,
{
eval: true,
},
);
assertEquals((await once(worker, "message"))[0], "It works!");
worker.terminate();
},
});
Deno.test({
name: "[node/worker_threads] Worker eval",
async fn() {
// Check that newlines are encoded properly
const worker = new workerThreads.Worker(
`
import { parentPort } from "node:worker_threads"
console.log("hey, foo") // comment
parentPort.postMessage("It works!");
`,
{
eval: true,
},
);
assertEquals((await once(worker, "message"))[0], "It works!");
worker.terminate();
},
});
Deno.test({
name: "[node/worker_threads] worker thread with type module",
async fn() {
function p() {
return new Promise<workerThreads.Worker>((resolve, reject) => {
const worker = new workerThreads.Worker(
new URL("./testdata/worker_module/index.js", import.meta.url),
);
worker.on("error", (e) => reject(e.message));
worker.on("message", () => resolve(worker));
});
}
await p();
},
});
Deno.test({
name: "[node/worker_threads] worker thread in nested module",
async fn() {
function p() {
return new Promise<workerThreads.Worker>((resolve, reject) => {
const worker = new workerThreads.Worker(
new URL("./testdata/worker_module/nested/index.js", import.meta.url),
);
worker.on("error", (e) => reject(e.message));
worker.on("message", () => resolve(worker));
});
}
await p();
},
});
Deno.test({
name: "[node/worker_threads] .cjs worker file within module",
async fn() {
function p() {
return new Promise<workerThreads.Worker>((resolve, reject) => {
const worker = new workerThreads.Worker(
new URL("./testdata/worker_module/cjs-file.cjs", import.meta.url),
);
worker.on("error", (e) => reject(e.message));
worker.on("message", () => resolve(worker));
});
}
await p();
},
});
Deno.test({
name: "[node/worker_threads] relativ path string",
async fn() {
function p() {
return new Promise<workerThreads.Worker>((resolve, reject) => {
const worker = new workerThreads.Worker(
"./tests/unit_node/testdata/worker_module/index.js",
);
worker.on("error", (e) => reject(e.message));
worker.on("message", () => resolve(worker));
});
}
await p();
},
});
Deno.test({
name: "[node/worker_threads] utf-8 path string",
async fn() {
function p() {
return new Promise<workerThreads.Worker>((resolve, reject) => {
const worker = new workerThreads.Worker(
"./tests/unit_node/testdata/worker_module/βάρβαροι.js",
);
worker.on("error", (e) => reject(e.message));
worker.on("message", () => resolve(worker));
});
}
await p();
},
});
Deno.test({
name: "[node/worker_threads] utf-8 path URL",
async fn() {
function p() {
return new Promise<workerThreads.Worker>((resolve, reject) => {
const worker = new workerThreads.Worker(
new URL(
"./testdata/worker_module/βάρβαροι.js",
import.meta.url,
),
);
worker.on("error", (e) => reject(e.message));
worker.on("message", () => resolve(worker));
});
}
await p();
},
});
Deno.test({
name: "[node/worker_threads] throws on relativ path without leading dot",
fn() {
assertThrows(
() => {
new workerThreads.Worker(
"tests/unit_node/testdata/worker_module/index.js",
);
},
);
},
});
Deno.test({
name: "[node/worker_threads] throws on unsupported URL protcol",
fn() {
assertThrows(
() => {
new workerThreads.Worker(new URL("https://example.com"));
},
);
},
});
Deno.test({
name: "[node/worker_threads] throws on non-existend file",
fn() {
assertThrows(
() => {
new workerThreads.Worker(new URL("file://very/unlikely"));
},
);
},
});
Deno.test({
name: "[node/worker_threads] inheritances",
async fn() {
const worker = new workerThreads.Worker(
`
import { EventEmitter } from "node:events";
import { parentPort } from "node:worker_threads";
parentPort.postMessage(parentPort instanceof EventTarget);
await new Promise(resolve => setTimeout(resolve, 100));
parentPort.postMessage(parentPort instanceof EventEmitter);
`,
{
eval: true,
},
);
assertEquals((await once(worker, "message"))[0], true);
assertEquals((await once(worker, "message"))[0], false);
assert(worker instanceof EventEmitter);
assert(!(worker instanceof EventTarget));
worker.terminate();
},
});
Deno.test({
name: "[node/worker_threads] Worker workerData",
async fn() {
const worker = new workerThreads.Worker(
new URL("./testdata/worker_threads.mjs", import.meta.url),
{
workerData: null,
},
);
worker.postMessage("Hello, how are you my thread?");
await once(worker, "message");
assertEquals((await once(worker, "message"))[0].workerData, null);
worker.terminate();
const worker1 = new workerThreads.Worker(
new URL("./testdata/worker_threads.mjs", import.meta.url),
);
worker1.postMessage("Hello, how are you my thread?");
await once(worker1, "message");
assertEquals((await once(worker1, "message"))[0].workerData, undefined);
worker1.terminate();
},
});
Deno.test({
name: "[node/worker_threads] Worker with relative path",
async fn() {
const worker = new workerThreads.Worker(
`.${SEPARATOR}` + relative(
Deno.cwd(),
fromFileUrl(new URL("./testdata/worker_threads.mjs", import.meta.url)),
),
);
worker.postMessage("Hello, how are you my thread?");
assertEquals((await once(worker, "message"))[0], "I'm fine!");
worker.terminate();
},
});
Deno.test({
name: "[node/worker_threads] unref",
async fn() {
const timeout = setTimeout(() => fail("Test timed out"), 60_000);
const child = new Deno.Command(Deno.execPath(), {
args: [
"eval",
"import { Worker } from 'node:worker_threads'; new Worker('setTimeout(() => {}, 1_000_000)', {eval:true}).unref();",
],
}).spawn();
await child.status;
clearTimeout(timeout);
},
});
Deno.test({
name: "[node/worker_threads] SharedArrayBuffer",
async fn() {
const sab = new SharedArrayBuffer(Uint8Array.BYTES_PER_ELEMENT);
const uint = new Uint8Array(sab);
const worker = new workerThreads.Worker(
new URL("./testdata/worker_threads2.mjs", import.meta.url),
{
workerData: { sharedArrayBuffer: sab },
},
);
worker.postMessage("Hello");
if ((await once(worker, "message"))[0] != "Hello") throw new Error();
await new Promise((resolve) => setTimeout(resolve, 100));
worker.terminate();
if (uint[0] != 1) throw new Error();
},
sanitizeResources: false,
});
Deno.test({
name: "[node/worker_threads] Worker workerData with MessagePort",
async fn() {
const { port1: mainPort, port2: workerPort } = new workerThreads
.MessageChannel();
const deferred = Promise.withResolvers<void>();
const worker = new workerThreads.Worker(
`
import {
isMainThread,
MessageChannel,
parentPort,
receiveMessageOnPort,
Worker,
workerData,
} from "node:worker_threads";
parentPort.on("message", (msg) => {
/* console.log("message from main", msg); */
parentPort.postMessage("Hello from worker on parentPort!");
workerData.workerPort.postMessage("Hello from worker on workerPort!");
});
`,
{
eval: true,
workerData: { workerPort },
transferList: [workerPort],
},
);
worker.on("message", (data) => {
assertEquals(data, "Hello from worker on parentPort!");
// TODO(bartlomieju): it would be better to use `mainPort.on("message")`,
// but we currently don't support it.
// https://github.com/denoland/deno/issues/22951
// Wait a bit so the message can arrive.
setTimeout(() => {
const msg = workerThreads.receiveMessageOnPort(mainPort)!.message;
assertEquals(msg, "Hello from worker on workerPort!");
deferred.resolve();
}, 500);
});
worker.postMessage("Hello from parent");
await deferred.promise;
await worker.terminate();
mainPort.close();
},
});
// Regression test for https://github.com/denoland/deno/issues/23362
Deno.test("[node/worker_threads] receiveMessageOnPort works if there's pending read", function () {
const { port1, port2 } = new workerThreads.MessageChannel();
const { port1: port3, port2: port4 } = new workerThreads.MessageChannel();
const { port1: port5, port2: port6 } = new workerThreads.MessageChannel();
const message1 = { hello: "world" };
const message2 = { foo: "bar" };
assertEquals(workerThreads.receiveMessageOnPort(port2), undefined);
port2.start();
port4.start();
port6.start();
port1.postMessage(message1);
port1.postMessage(message2);
port3.postMessage(message1);
port3.postMessage(message2);
port5.postMessage(message1);
port5.postMessage(message2);
assertEquals(workerThreads.receiveMessageOnPort(port2), {
message: message1,
});
assertEquals(workerThreads.receiveMessageOnPort(port2), {
message: message2,
});
assertEquals(workerThreads.receiveMessageOnPort(port4), {
message: message1,
});
assertEquals(workerThreads.receiveMessageOnPort(port4), {
message: message2,
});
assertEquals(workerThreads.receiveMessageOnPort(port6), {
message: message1,
});
assertEquals(workerThreads.receiveMessageOnPort(port6), {
message: message2,
});
port1.close();
port2.close();
port3.close();
port4.close();
port5.close();
port6.close();
});
Deno.test({
name: "[node/worker_threads] Worker env",
async fn() {
const deferred = Promise.withResolvers<void>();
const worker = new workerThreads.Worker(
`
import { parentPort } from "node:worker_threads";
import process from "node:process";
parentPort.postMessage(process.env.TEST_ENV);
`,
{
eval: true,
env: { TEST_ENV: "test" },
},
);
worker.on("message", (data) => {
assertEquals(data, "test");
deferred.resolve();
});
await deferred.promise;
await worker.terminate();
},
});
Deno.test({
name: "[node/worker_threads] Worker env using process.env",
async fn() {
const deferred = Promise.withResolvers<void>();
const worker = new workerThreads.Worker(
`
import { parentPort } from "node:worker_threads";
import process from "node:process";
parentPort.postMessage("ok");
`,
{
eval: true,
// Make sure this doesn't throw `DataCloneError`.
// See https://github.com/denoland/deno/issues/23522.
env: process.env,
},
);
worker.on("message", (data) => {
assertEquals(data, "ok");
deferred.resolve();
});
await deferred.promise;
await worker.terminate();
},
});
Deno.test({
name: "[node/worker_threads] Returns terminate promise with exit code",
async fn() {
const deferred = Promise.withResolvers<void>();
const worker = new workerThreads.Worker(
`
import { parentPort } from "node:worker_threads";
parentPort.postMessage("ok");
`,
{
eval: true,
},
);
worker.on("message", (data) => {
assertEquals(data, "ok");
deferred.resolve();
});
await deferred.promise;
const promise = worker.terminate();
assertEquals(typeof promise.then, "function");
assertEquals(await promise, 0);
},
});
Deno.test({
name:
"[node/worker_threads] MessagePort.on all message listeners are invoked",
async fn() {
const output: string[] = [];
const deferred = Promise.withResolvers<void>();
const { port1, port2 } = new workerThreads.MessageChannel();
port1.on("message", (msg) => output.push(msg));
port1.on("message", (msg) => output.push(msg + 2));
port1.on("message", (msg) => {
output.push(msg + 3);
deferred.resolve();
});
port2.postMessage("hi!");
await deferred.promise;
assertEquals(output, ["hi!", "hi!2", "hi!3"]);
port2.close();
port1.close();
},
});
// Test for https://github.com/denoland/deno/issues/23854
Deno.test({
name: "[node/worker_threads] MessagePort.addListener is present",
async fn() {
const channel = new workerThreads.MessageChannel();
const worker = new workerThreads.Worker(
`
import { parentPort } from "node:worker_threads";
parentPort.addListener("message", message => {
if (message.foo) {
const success = typeof message.foo.bar.addListener === "function";
parentPort.postMessage(success ? "it works" : "it doesn't work")
}
})
`,
{
eval: true,
},
);
worker.postMessage({ foo: { bar: channel.port1 } }, [channel.port1]);
assertEquals((await once(worker, "message"))[0], "it works");
worker.terminate();
channel.port1.close();
channel.port2.close();
},
});
Deno.test({
name: "[node/worker_threads] Emits online event",
async fn() {
const worker = new workerThreads.Worker(
`
import { parentPort } from "node:worker_threads";
const p = Promise.withResolvers();
let ok = false;
parentPort.on("message", () => {
ok = true;
p.resolve();
});
await Promise.race([p.promise, new Promise(resolve => setTimeout(resolve, 20000))]);
if (ok) {
parentPort.postMessage("ok");
} else {
parentPort.postMessage("timed out");
}
`,
{
eval: true,
},
);
worker.on("online", () => {
worker.postMessage("ok");
});
assertEquals((await once(worker, "message"))[0], "ok");
worker.terminate();
},
});
Deno.test({
name: "[node/worker_threads] receiveMessageOnPort doesn't exit receive loop",
async fn() {
const worker = new workerThreads.Worker(
`
import { parentPort, receiveMessageOnPort } from "node:worker_threads";
parentPort.on("message", (msg) => {
const port = msg.port;
port.on("message", (msg2) => {
if (msg2 === "c") {
port.postMessage("done");
port.unref();
parentPort.unref();
}
});
parentPort.postMessage("ready");
const msg2 = receiveMessageOnPort(port);
});
`,
{ eval: true },
);
const { port1, port2 } = new workerThreads.MessageChannel();
worker.postMessage({ port: port2 }, [port2]);
const done = Promise.withResolvers<boolean>();
port1.on("message", (msg) => {
assertEquals(msg, "done");
worker.unref();
port1.close();
done.resolve(true);
});
worker.on("message", (msg) => {
assertEquals(msg, "ready");
port1.postMessage("a");
port1.postMessage("b");
port1.postMessage("c");
});
const timeout = setTimeout(() => {
fail("Test timed out");
}, 20_000);
try {
const result = await done.promise;
assertEquals(result, true);
} finally {
clearTimeout(timeout);
}
},
});
Deno.test({
name: "[node/worker_threads] MessagePort.unref doesn't exit receive loop",
async fn() {
const worker = new workerThreads.Worker(
`
import { parentPort } from "node:worker_threads";
const assertEquals = (a, b) => {
if (a !== b) {
throw new Error();
}
};
let state = 0;
parentPort.on("message", (msg) => {
const port = msg.port;
const expect = ["a", "b", "c"];
port.on("message", (msg2) => {
assertEquals(msg2, expect[state++]);
if (msg2 === "c") {
port.postMessage({ type: "done", got: msg2 });
parentPort.unref();
}
});
port.unref();
parentPort.postMessage("ready");
});
`,
{ eval: true },
);
const { port1, port2 } = new workerThreads.MessageChannel();
const done = Promise.withResolvers<boolean>();
port1.on("message", (msg) => {
assertEquals(msg.type, "done");
assertEquals(msg.got, "c");
worker.unref();
port1.close();
done.resolve(true);
});
worker.on("message", (msg) => {
assertEquals(msg, "ready");
port1.postMessage("a");
port1.postMessage("b");
port1.postMessage("c");
});
worker.postMessage({ port: port2 }, [port2]);
const timeout = setTimeout(() => {
fail("Test timed out");
}, 20_000);
try {
const result = await done.promise;
assertEquals(result, true);
} finally {
clearTimeout(timeout);
}
},
});
Deno.test({
name: "[node/worker_threads] npm:piscina wait loop hang regression",
async fn() {
const worker = new workerThreads.Worker(
`
import { assert, assertEquals } from "@std/assert";
import { parentPort, receiveMessageOnPort } from "node:worker_threads";
assert(parentPort !== null);
let currentTasks = 0;
let lastSeen = 0;
parentPort.on("message", (msg) => {
(async () => {
assert(typeof msg === "object" && msg !== null);
assert(msg.buf !== undefined);
assert(msg.port !== undefined);
const { buf, port } = msg;
port.postMessage("ready");
port.on("message", (msg) => onMessage(msg, buf, port));
atomicsWaitLoop(buf, port);
})();
});
function onMessage(msg, buf, port) {
currentTasks++;
(async () => {
assert(msg.taskName !== undefined);
port.postMessage({ type: "response", taskName: msg.taskName });
currentTasks--;
atomicsWaitLoop(buf, port);
})();
}
function atomicsWaitLoop(buf, port) {
while (currentTasks === 0) {
Atomics.wait(buf, 0, lastSeen);
lastSeen = Atomics.load(buf, 0);
let task;
while ((task = receiveMessageOnPort(port)) !== undefined) {
onMessage(task.message, buf, port);
}
}
}
`,
{ eval: true },
);
const sab = new SharedArrayBuffer(4);
const buf = new Int32Array(sab);
const { port1, port2 } = new workerThreads.MessageChannel();
const done = Promise.withResolvers<boolean>();
port1.unref();
worker.postMessage({
type: "init",
buf,
port: port2,
}, [port2]);
let count = 0;
port1.on("message", (msg) => {
if (count++ === 0) {
assertEquals(msg, "ready");
} else {
assertEquals(msg.type, "response");
port1.close();
done.resolve(true);
}
});
port1.postMessage({
taskName: "doThing",
});
Atomics.add(buf, 0, 1);
Atomics.notify(buf, 0, 1);
worker.unref();
const result = await done.promise;
assertEquals(result, true);
},
});