feat: worker_threads

- fix bug for when there was multiple workers defined in the same file
- added a new function that behave like an async function but run in parallel
- added a new helper to execute a function in background

Author: MP281X

stdlib/index.ts

@@ -1,5 +1,7 @@
 export { Schema } from '@effect/schema'
 
+export * from './lib/execParallel.ts'
+export * from './lib/executionOrder.ts'
 export * from './lib/pubSub.ts'
 export * from './lib/queue.ts'
 export * from './lib/test.ts'

stdlib/lib/execParallel.ts

@@ -0,0 +1,32 @@
+import { fileURLToPath } from 'node:url'
+import { Worker, isMainThread, parentPort, workerData } from 'node:worker_threads'
+import { getExecutionId } from './executionOrder.ts'
+import { fork, resolvablePromise } from './utils.ts'
+
+export function execParallel<Props, ReturnType>(filePath: string, fn: (props: Props) => ReturnType) {
+	const { executionId } = getExecutionId(filePath)
+
+	if (isMainThread && process.env['execParallelId'] === executionId) {
+		fork(async () => {
+			const res = await fn(workerData)
+			parentPort?.postMessage(res)
+
+			process.exit(0)
+		})
+	}
+
+	return (props: Props) => {
+		const worker = new Worker(fileURLToPath(filePath), {
+			workerData: props,
+			env: { ...process.env, execParallelId: executionId }
+		})
+
+		const { promise, resolvePromise } = resolvablePromise<ReturnType>()
+
+		worker.on('message', data => {
+			resolvePromise(data)
+		})
+
+		return promise
+	}
+}

stdlib/lib/executionOrder.ts

@@ -0,0 +1,14 @@
+const executionMap = new Map<string, number>()
+
+/**
+ * Return an Id that it's consistent for every execution
+ * It's based on the order of execution
+ */
+export const getExecutionId = (fileName: string) => {
+	executionMap.get(fileName) ?? executionMap.set(fileName, 0)
+
+	const index = executionMap.get(fileName) ?? 0
+	executionMap.set(fileName, index + 1)
+
+	return { executionId: index.toString() }
+}

stdlib/lib/utils.ts

@@ -30,6 +30,14 @@ export function resolvablePromise<T>() {
 	return { resolvePromise, promise }
 }
 
+export async function fork<ReturnType extends Promise<any> | any>(fn: (index: number) => ReturnType, instances = 1) {
+	const promises: ReturnType[] = []
+
+	for (let i = 0; i < instances; i++) promises.push(fn(i))
+
+	return await Promise.all(promises)
+}
+
 export declare namespace random {
 	type Props = { min: number; max: number; step: number }
 }

stdlib/lib/workerThreads.ts

@@ -1,6 +1,7 @@
 import { fileURLToPath } from 'node:url'
 import { Worker, isMainThread, parentPort, workerData as rawWorkerData } from 'node:worker_threads'
 import { Schema } from '@effect/schema'
+import { getExecutionId } from './executionOrder.ts'
 import { Queue } from './queue.ts'
 
 export declare namespace workerThreads {
@@ -25,49 +26,57 @@ export declare namespace workerThreads {
 }
 
 export async function workerThreads<Schema extends workerThreads.Schema>(props: workerThreads.Props<Schema>) {
-	if (isMainThread) {
-		const spawnWorker = (workerData: Schema['init']['Encoded']) => {
-			const worker = new Worker(fileURLToPath(props.filePath), { workerData })
+	const { executionId } = getExecutionId(props.filePath)
 
-			const queue = new Queue<Schema['main']['Type']>()
-			worker.on('message', rawData => {
-				const data = Schema.decodeSync(props.schema.main)(rawData)
-				void queue.publish(data)
-			})
+	if (isMainThread && process.env['workerThreadsId'] === executionId) {
+		const worker = parentPort!
+		const workerData = Schema.decodeSync(props.schema.init)(rawWorkerData)
 
-			worker.on('exit', () => queue.close())
-			worker.on('error', () => queue.close())
-			worker.on('messageerror', () => queue.close())
+		const queue = new Queue<Schema['worker']['Type']>()
+		worker.on('message', rawData => {
+			const data = Schema.decodeSync(props.schema.worker)(rawData)
+			void queue.publish(data)
+		})
 
-			const send = async (rawData: Schema['worker']['Encoded']) => {
-				const data = await Schema.encodePromise(props.schema.worker)(rawData)
-				worker.postMessage(data)
-			}
+		worker.on('exit', () => queue.close())
+		worker.on('error', () => queue.close())
+		worker.on('messageerror', () => queue.close())
 
-			return { queue, send, terminate: worker.terminate }
+		const send = async (rawData: Schema['main']['Encoded']) => {
+			const data = await Schema.encodePromise(props.schema.main)(rawData)
+			worker.postMessage(data)
 		}
 
-		return { spawnWorker }
+		await props.worker({ queue, send, workerData, terminate: process.exit })
+		process.exit(0)
 	}
 
-	const worker = parentPort!
-	const workerData = Schema.decodeSync(props.schema.init)(rawWorkerData)
+	function spawnWorker(workerData: Schema['init']['Encoded']) {
+		const worker = new Worker(fileURLToPath(props.filePath), {
+			workerData,
+			env: {
+				...process.env,
+				workerThreadsId: executionId
+			}
+		})
+
+		const queue = new Queue<Schema['main']['Type']>()
+		worker.on('message', rawData => {
+			const data = Schema.decodeSync(props.schema.main)(rawData)
+			void queue.publish(data)
+		})
 
-	const queue = new Queue<Schema['worker']['Type']>()
-	worker.on('message', rawData => {
-		const data = Schema.decodeSync(props.schema.worker)(rawData)
-		void queue.publish(data)
-	})
+		worker.on('exit', () => queue.close())
+		worker.on('error', () => queue.close())
+		worker.on('messageerror', () => queue.close())
 
-	worker.on('exit', () => queue.close())
-	worker.on('error', () => queue.close())
-	worker.on('messageerror', () => queue.close())
+		const send = async (rawData: Schema['worker']['Encoded']) => {
+			const data = await Schema.encodePromise(props.schema.worker)(rawData)
+			worker.postMessage(data)
+		}
 
-	const send = async (rawData: Schema['main']['Encoded']) => {
-		const data = await Schema.encodePromise(props.schema.main)(rawData)
-		worker.postMessage(data)
+		return { queue, send, terminate: worker.terminate }
 	}
 
-	await props.worker({ queue, send, workerData, terminate: process.exit })
-	process.exit(0)
+	return { spawnWorker }
 }