Try creating a new worker for each request
This commit is contained in:
@@ -1,6 +1,6 @@
|
|||||||
{
|
{
|
||||||
"compilerOptions": {
|
"compilerOptions": {
|
||||||
"lib": ["deno.window", "deno.worker"]
|
"lib": ["deno.window", "deno.worker", "deno.ns"]
|
||||||
},
|
},
|
||||||
"lint": {
|
"lint": {
|
||||||
"files": {
|
"files": {
|
||||||
|
|||||||
@@ -6,13 +6,13 @@ const States = Object.freeze({
|
|||||||
TIMEOUT: "TIMEOUT",
|
TIMEOUT: "TIMEOUT",
|
||||||
});
|
});
|
||||||
|
|
||||||
// Create a pool of workers.
|
async function handler(req: Request): Promise<Response> {
|
||||||
const workerPool: { worker: Worker; inUse: boolean }[] = [];
|
if (req.method !== "POST" || new URL(req.url).pathname !== "/execute") {
|
||||||
// Allow the number of workers to be configured via an environment variable.
|
return new Response("Not Found", { status: 404 });
|
||||||
const numWorkers =
|
}
|
||||||
parseInt(Deno.env.get("NUM_WORKERS") || "0") || navigator.hardwareConcurrency;
|
|
||||||
|
|
||||||
for (let i = 0; i < numWorkers; i++) {
|
// For each request, spawn a new worker, use it, and then terminate it.
|
||||||
|
// This ensures a clean, stateless environment for every execution.
|
||||||
const worker = new Worker(new URL("./worker.ts", import.meta.url).href, {
|
const worker = new Worker(new URL("./worker.ts", import.meta.url).href, {
|
||||||
type: "module",
|
type: "module",
|
||||||
deno: {
|
deno: {
|
||||||
@@ -20,47 +20,6 @@ for (let i = 0; i < numWorkers; i++) {
|
|||||||
permissions: "inherit",
|
permissions: "inherit",
|
||||||
},
|
},
|
||||||
});
|
});
|
||||||
workerPool.push({ worker, inUse: false });
|
|
||||||
}
|
|
||||||
|
|
||||||
const requestQueue: ((
|
|
||||||
value:
|
|
||||||
| { worker: Worker; inUse: boolean }
|
|
||||||
| PromiseLike<{ worker: Worker; inUse: boolean }>
|
|
||||||
) => void)[] = [];
|
|
||||||
|
|
||||||
function getAvailableWorker() {
|
|
||||||
return new Promise((resolve) => {
|
|
||||||
const availableWorker = workerPool.find((w) => !w.inUse);
|
|
||||||
if (availableWorker) {
|
|
||||||
availableWorker.inUse = true;
|
|
||||||
resolve(availableWorker);
|
|
||||||
} else {
|
|
||||||
requestQueue.push(resolve);
|
|
||||||
}
|
|
||||||
});
|
|
||||||
}
|
|
||||||
|
|
||||||
function releaseWorker(worker) {
|
|
||||||
if (requestQueue.length > 0) {
|
|
||||||
const nextRequest = requestQueue.shift();
|
|
||||||
if (nextRequest) {
|
|
||||||
nextRequest(worker);
|
|
||||||
}
|
|
||||||
} else {
|
|
||||||
worker.inUse = false;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
async function handler(req: Request): Promise<Response> {
|
|
||||||
if (req.method !== "POST" || new URL(req.url).pathname !== "/execute") {
|
|
||||||
return new Response("Not Found", { status: 404 });
|
|
||||||
}
|
|
||||||
|
|
||||||
const availableWorker = (await getAvailableWorker()) as {
|
|
||||||
worker: Worker;
|
|
||||||
inUse: boolean;
|
|
||||||
};
|
|
||||||
|
|
||||||
try {
|
try {
|
||||||
const body = await req.json();
|
const body = await req.json();
|
||||||
@@ -75,8 +34,8 @@ async function handler(req: Request): Promise<Response> {
|
|||||||
let timeoutId: number;
|
let timeoutId: number;
|
||||||
|
|
||||||
const cleanup = () => {
|
const cleanup = () => {
|
||||||
availableWorker.worker.removeEventListener("message", messageHandler);
|
worker.removeEventListener("message", messageHandler);
|
||||||
availableWorker.worker.removeEventListener("error", errorHandler);
|
worker.removeEventListener("error", errorHandler);
|
||||||
if (timeoutId) clearTimeout(timeoutId);
|
if (timeoutId) clearTimeout(timeoutId);
|
||||||
};
|
};
|
||||||
|
|
||||||
@@ -95,10 +54,10 @@ async function handler(req: Request): Promise<Response> {
|
|||||||
reject(new Error("Timeout"));
|
reject(new Error("Timeout"));
|
||||||
}, TIMEOUT_MS);
|
}, TIMEOUT_MS);
|
||||||
|
|
||||||
availableWorker.worker.addEventListener("message", messageHandler);
|
worker.addEventListener("message", messageHandler);
|
||||||
availableWorker.worker.addEventListener("error", errorHandler);
|
worker.addEventListener("error", errorHandler);
|
||||||
|
|
||||||
availableWorker.worker.postMessage({ code, request, environment, name });
|
worker.postMessage({ code, request, environment, name });
|
||||||
});
|
});
|
||||||
|
|
||||||
const startTime = performance.now();
|
const startTime = performance.now();
|
||||||
@@ -124,12 +83,12 @@ async function handler(req: Request): Promise<Response> {
|
|||||||
} catch (e) {
|
} catch (e) {
|
||||||
return new Response(`Bad Request: ${e.message}`, { status: 400 });
|
return new Response(`Bad Request: ${e.message}`, { status: 400 });
|
||||||
} finally {
|
} finally {
|
||||||
releaseWorker(availableWorker); // Release the worker.
|
worker.terminate(); // Ensure the worker is always terminated.
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
const port = parseInt(Deno.env.get("PORT") || "8000");
|
const port = parseInt(Deno.env.get("PORT") || "8000");
|
||||||
|
|
||||||
console.log(`⚡ Deno server ready on :${port} with ${numWorkers} workers.`);
|
console.log(`⚡ Deno server ready on :${port}`);
|
||||||
|
|
||||||
await serve(handler, { port });
|
await serve(handler, { port });
|
||||||
|
|||||||
Reference in New Issue
Block a user