feat: worker pooling

Cargo.toml

@@ -24,7 +24,7 @@ exclude = [
 ]
 
 [workspace.package]
-version = "0.5.0"
+version = "0.5.1-rc.0"
 edition = "2021"
 
 [workspace.dependencies]

docs/metatype.dev/docs/reference/typegate/index.mdx

@@ -82,3 +82,9 @@ The following environment variables can be used to configure the typegate. `SYNC
 | SUBSTANTIAL_POLL_INTERVAL_SEC    | Rate at which new schedules are read.                                                                    | 1.0                      | 0.6                                                                                      |
 | SUBSTANTIAL_LEASE_LIFESPAN_SEC   | Lease duration associated to a workflow run                                                              | 2.0                      | 6                                                                                        |
 | SUBSTANTIAL_MAX_ACQUIRE_PER_TICK | Max amount of new acquired replay requests per tick                                                      | 3                        | 5                                                                                        |
+| MIN_DENO_WORKERS | Minimal number of available deno workers | 2 | 4 |
+| MAX_DENO_WORKERS | Maximal number of available deno workers | 8 | 16 |
+| DENO_WORKER_WAIT_TIMEOUT_MS | Timeout for waiting for a free deno worker | 5000 | 2000 |
+| MIN_SUBSTANTIAL_WORKERS | Minimal number of available substantial workers | 2 | 4 |
+| MAX_SUBSTANTIAL_WORKERS | Maximal number of available substantial workers | 8 | 16 |
+| SUBSTANTIAL_WORKER_WAIT_TIMEOUT_MS | Timeout for waiting for a free substantial worker | 15000 | 2000 |

examples/templates/deno/api/example.ts

@@ -1,6 +1,6 @@
-import { Policy, t, typegraph } from "jsr:@typegraph/[email protected]";
-import { PythonRuntime } from "jsr:@typegraph/[email protected]/runtimes/python";
-import { DenoRuntime } from "jsr:@typegraph/[email protected]/runtimes/deno";
+import { Policy, t, typegraph } from "jsr:@typegraph/[email protected].1-rc.0";
+import { PythonRuntime } from "jsr:@typegraph/[email protected].1-rc.0/runtimes/python";
+import { DenoRuntime } from "jsr:@typegraph/[email protected].1-rc.0/runtimes/deno";
 
 await typegraph("example", (g) => {
   const pub = Policy.public();

examples/templates/deno/compose.yml

@@ -1,6 +1,6 @@
 services:
   typegate:
-    image: ghcr.io/metatypedev/typegate:v0.5.0
+    image: ghcr.io/metatypedev/typegate:v0.5.1-rc.0
     restart: always
     ports:
       - "7890:7890"

examples/templates/node/compose.yml

@@ -1,6 +1,6 @@
 services:
   typegate:
-    image: ghcr.io/metatypedev/typegate:v0.5.0
+    image: ghcr.io/metatypedev/typegate:v0.5.1-rc.0
     restart: always
     ports:
       - "7890:7890"

examples/templates/node/package.json

@@ -6,7 +6,7 @@
     "dev": "MCLI_LOADER_CMD='npm x tsx' meta dev"
   },
   "dependencies": {
-    "@typegraph/sdk": "^0.5.0"
+    "@typegraph/sdk": "^0.5.1-rc.0"
   },
   "devDependencies": {
     "tsx": "^3.13.0",

examples/templates/python/compose.yml

@@ -1,6 +1,6 @@
 services:
   typegate:
-    image: ghcr.io/metatypedev/typegate:v0.5.0
+    image: ghcr.io/metatypedev/typegate:v0.5.1-rc.0
     restart: always
     ports:
       - "7890:7890"

examples/templates/python/pyproject.toml

@@ -1,12 +1,12 @@
 [tool.poetry]
 name = "example"
-version = "0.5.0"
+version = "0.5.1-rc.0"
 description = ""
 authors = []
 
 [tool.poetry.dependencies]
 python = ">=3.8,<4.0"
-typegraph = "0.5.0"
+typegraph = "0.5.1-rc.0"
 
 [build-system]
 requires = ["poetry-core"]

pyproject.toml

@@ -6,7 +6,7 @@
 
 [tool.poetry]
 name = "metatype"
-version = "0.5.0"
+version = "0.5.1-rc.0"
 description = ""
 authors = []
 

src/pyrt_wit_wire/pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "pyrt_wit_wire"
-version = "0.5.0"
+version = "0.5.1-rc.0"
 description = "Wasm component implementing the PythonRuntime host using wit_wire protocol."
 license = "MPL-2.0"
 readme = "README.md"

src/typegate/engine/src/runtimes/wit_wire.rs

@@ -405,7 +405,7 @@ pub fn op_wit_wire_destroy(
     scope: &mut v8::HandleScope<'_>,
     #[string] instance_id: String,
 ) {
-    debug!("destroying wit_wire instnace {instance_id}");
+    debug!("destroying wit_wire instance {instance_id}");
     let ctx = {
         let state = state.borrow();
         let ctx = state.borrow::<Ctx>();

src/typegate/src/config/types.ts

@@ -60,6 +60,16 @@ export const globalConfigSchema = z.object({
   sentry_sample_rate: z.coerce.number().positive().min(0).max(1),
   sentry_traces_sample_rate: z.coerce.number().positive().min(0).max(1),
   deno_v8_flags: z.string().optional(),
+  min_deno_workers: z.coerce.number().positive().default(2),
+  max_deno_workers: z.coerce.number().positive().default(8),
+  deno_worker_wait_timeout_ms: z.coerce.number().positive().default(5000),
+  // deno_idle_worker_timeout_ms: z.coerce.number().positive().optional(), // auto remove idle workers
+  min_substantial_workers: z.coerce.number().positive().default(2),
+  max_substantial_workers: z.coerce.number().positive().default(8),
+  substantial_worker_wait_timeout_ms: z.coerce.number().positive().default(
+    15000,
+  ),
+  // substantial_idle_worker_timeout_ms: z.coerce.number().positive().optional(), // auto remove idle workers
 });
 export type GlobalConfig = z.infer<typeof globalConfigSchema>;
 
@@ -122,7 +132,7 @@ export type SyncConfigX = {
   redis: RedisConnectOptions;
   s3: S3ClientConfig;
   s3Bucket: string;
-  forceRemove?: boolean
+  forceRemove?: boolean;
 };
 
 export type TypegateConfig = {

src/typegate/src/runtimes/deno/deno.ts

@@ -182,7 +182,7 @@ export class DenoRuntime extends Runtime {
   }
 
   async deinit(): Promise<void> {
-    // await this.workerManager.deinit();
+    this.workerManager.deinit();
   }
 
   materialize(

src/typegate/src/runtimes/deno/worker_manager.ts

@@ -1,12 +1,14 @@
 // Copyright Metatype OÜ, licensed under the Mozilla Public License Version 2.0.
 // SPDX-License-Identifier: MPL-2.0
 
+import { globalConfig } from "../../config.ts";
 import { getLogger } from "../../log.ts";
 import { DenoWorker } from "../patterns/worker_manager/deno.ts";
 import {
   BaseWorkerManager,
   createTaskId,
 } from "../patterns/worker_manager/mod.ts";
+import { WorkerPool } from "../patterns/worker_manager/pooling.ts";
 import { TaskId } from "../patterns/worker_manager/types.ts";
 import { TaskContext } from "./shared_types.ts";
 import { DenoEvent, DenoMessage, TaskSpec } from "./types.ts";
@@ -19,23 +21,35 @@ export type WorkerManagerConfig = {
 
 export class WorkerManager
   extends BaseWorkerManager<TaskSpec, DenoMessage, DenoEvent> {
+  static #pool: WorkerPool<TaskSpec, DenoMessage, DenoEvent> | null = null;
+  static #getPool() {
+    if (!WorkerManager.#pool) {
+      WorkerManager.#pool = new WorkerPool(
+        "deno runtime",
+        {
+          minWorkers: globalConfig.min_deno_workers,
+          maxWorkers: globalConfig.max_deno_workers,
+          waitTimeoutMs: globalConfig.deno_worker_wait_timeout_ms,
+        },
+        (id: string) => new DenoWorker(id, import.meta.resolve("./worker.ts")),
+      );
+    }
+    return WorkerManager.#pool!;
+  }
+
   constructor(private config: WorkerManagerConfig) {
-    super(
-      (taskId: TaskId) => {
-        return new DenoWorker(taskId, import.meta.resolve("./worker.ts"));
-      },
-    );
+    super(WorkerManager.#getPool());
   }
 
-  callFunction(
+  async callFunction(
     name: string,
     modulePath: string,
     relativeModulePath: string,
     args: unknown,
     internalTCtx: TaskContext,
   ) {
     const taskId = createTaskId(`${name}@${relativeModulePath}`);
-    this.createWorker(name, taskId, {
+    await this.delegateTask(name, taskId, {
       modulePath,
       functionName: name,
     });
@@ -49,13 +63,13 @@ export class WorkerManager
 
     return new Promise((resolve, reject) => {
       const timeoutId = setTimeout(() => {
-        this.destroyWorker(name, taskId);
+        this.deallocateWorker(name, taskId);
         reject(new Error(`${this.config.timeout_ms}ms timeout exceeded`));
       }, this.config.timeout_ms);
 
       const handler: (event: DenoEvent) => void = (event) => {
         clearTimeout(timeoutId);
-        this.destroyWorker(name, taskId);
+        this.deallocateWorker(name, taskId);
         switch (event.type) {
           case "SUCCESS":
             resolve(event.result);

src/typegate/src/runtimes/patterns/worker_manager/deno.ts

@@ -3,7 +3,7 @@
 
 import { envSharedWithWorkers } from "../../../config/shared.ts";
 import { BaseWorker } from "./mod.ts";
-import { BaseMessage, EventHandler, TaskId } from "./types.ts";
+import { BaseMessage, EventHandler } from "./types.ts";
 
 export interface DenoWorkerError extends BaseMessage {
   type: "WORKER_ERROR";
@@ -15,11 +15,11 @@ export type BaseDenoWorkerMessage = BaseMessage | DenoWorkerError;
 export class DenoWorker<M extends BaseMessage, E extends BaseDenoWorkerMessage>
   extends BaseWorker<M, E> {
   #worker: Worker;
-  #taskId: TaskId;
-  constructor(taskId: TaskId, workerPath: string) {
+  #workerId: string;
+  constructor(workerId: string, workerPath: string) {
     super();
     this.#worker = new Worker(workerPath, {
-      name: taskId,
+      name: workerId,
       type: "module",
       deno: {
         permissions: {
@@ -35,21 +35,22 @@ export class DenoWorker<M extends BaseMessage, E extends BaseDenoWorkerMessage>
         },
       },
     });
-    this.#taskId = taskId;
+    this.#workerId = workerId;
   }
 
   listen(handlerFn: EventHandler<E>) {
     this.#worker.onmessage = async (message) => {
       await handlerFn(message.data as E);
     };
 
-    this.#worker.onerror = /*async*/ (event) =>
-      handlerFn(
+    this.#worker.onerror = async (event) => {
+      await handlerFn(
         {
           type: "WORKER_ERROR",
           event,
         } as E,
       );
+    };
   }
 
   send(msg: M) {
@@ -61,6 +62,6 @@ export class DenoWorker<M extends BaseMessage, E extends BaseDenoWorkerMessage>
   }
 
   get id() {
-    return this.#taskId;
+    return this.#workerId;
   }
 }