diff --git a/examples/multi-module/main.ts b/examples/multi-module/main.ts index b3aa93f..32862e3 100644 --- a/examples/multi-module/main.ts +++ b/examples/multi-module/main.ts @@ -12,7 +12,7 @@ import { uppercase, repeat } from './text.ts'; using run = workers(1); // All four moroutines from two modules run on the same single worker - const results = await Promise.all([run(square(7)), run(add(3, 4)), run(uppercase('hello')), run(repeat('ab', 3))]); + const results = await run([square(7), add(3, 4), uppercase('hello'), repeat('ab', 3)]); console.log('square(7) =', results[0]); // 49 console.log('add(3, 4) =', results[1]); // 7 diff --git a/examples/non-blocking/main.ts b/examples/non-blocking/main.ts index fee8eac..93837d5 100644 --- a/examples/non-blocking/main.ts +++ b/examples/non-blocking/main.ts @@ -20,7 +20,7 @@ console.log('Meanwhile, the main thread keeps ticking:\n'); { using run = workers(2); const start = performance.now(); - const [a, b] = await Promise.all([run(fibonacci(42)), run(fibonacci(41))]); + const [a, b] = await run([fibonacci(42), fibonacci(41)]); const elapsed = (performance.now() - start).toFixed(0); clearInterval(interval); diff --git a/examples/shared-state/main.ts b/examples/shared-state/main.ts index 289e29d..cb39e27 100644 --- a/examples/shared-state/main.ts +++ b/examples/shared-state/main.ts @@ -16,11 +16,11 @@ const steps = 1000; using run = workers(4); // 4 workers each move the position (1, 2) per step, 1000 steps each - await Promise.all([ - run(updatePosition(mu, pos, 1, 2, steps)), - run(updatePosition(mu, pos, 1, 2, steps)), - run(updatePosition(mu, pos, 1, 2, steps)), - run(updatePosition(mu, pos, 1, 2, steps)), + await run([ + updatePosition(mu, pos, 1, 2, steps), + updatePosition(mu, pos, 1, 2, steps), + updatePosition(mu, pos, 1, 2, steps), + updatePosition(mu, pos, 1, 2, steps), ]); } diff --git a/src/runner.ts b/src/runner.ts index 2d8eed3..87009c7 100644 --- a/src/runner.ts +++ b/src/runner.ts @@ -1,6 +1,9 @@ import type { Task } from './task.ts'; +type TaskResults[]> = { [K in keyof T]: T[K] extends Task ? R : never }; + export type Runner = { (task: Task): Promise; + []>(tasks: [...T]): Promise>; [Symbol.dispose](): void; }; diff --git a/src/worker-pool.ts b/src/worker-pool.ts index c042d57..1c67a14 100644 --- a/src/worker-pool.ts +++ b/src/worker-pool.ts @@ -17,12 +17,19 @@ export function workers(size: number): Runner { let next = 0; let disposed = false; + function dispatch(task: Task): Promise { + if (disposed) return Promise.reject(new Error('Worker pool is disposed')); + const worker = pool[next % pool.length]; + next++; + return execute(worker, task.id, task.args); + } + const run: Runner = Object.assign( - (task: Task): Promise => { - if (disposed) return Promise.reject(new Error('Worker pool is disposed')); - const worker = pool[next % pool.length]; - next++; - return execute(worker, task.id, task.args); + (taskOrTasks: Task | Task[]): any => { + if (Array.isArray(taskOrTasks)) { + return Promise.all(taskOrTasks.map((t) => dispatch(t))); + } + return dispatch(taskOrTasks); }, { [Symbol.dispose]() { diff --git a/test/pool.test.ts b/test/pool.test.ts index 579d014..4375eb3 100644 --- a/test/pool.test.ts +++ b/test/pool.test.ts @@ -17,7 +17,7 @@ describe('workers', () => { it('handles concurrent calls across pool workers', async () => { const run = workers(2); try { - const results = await Promise.all([run(double(1)), run(double(2)), run(double(3)), run(double(4))]); + const results = await run([double(1), double(2), double(3), double(4)]); assert.deepEqual(results, [2, 4, 6, 8]); } finally { run[Symbol.dispose](); diff --git a/test/shared/cross-worker.test.ts b/test/shared/cross-worker.test.ts index 0908c21..b58c285 100644 --- a/test/shared/cross-worker.test.ts +++ b/test/shared/cross-worker.test.ts @@ -9,7 +9,7 @@ describe('shared primitives across workers', () => { const counter = int32atomic(); const run = workers(2); try { - await Promise.all([run(atomicAdd(counter, 10)), run(atomicAdd(counter, 20))]); + await run([atomicAdd(counter, 10), atomicAdd(counter, 20)]); assert.equal(counter.load(), 30); } finally { run[Symbol.dispose](); @@ -21,7 +21,7 @@ describe('shared primitives across workers', () => { const counter = int32atomic(); const run = workers(2); try { - await Promise.all([run(mutexIncrement(m, counter, 100)), run(mutexIncrement(m, counter, 100))]); + await run([mutexIncrement(m, counter, 100), mutexIncrement(m, counter, 100)]); assert.equal(counter.load(), 200); } finally { run[Symbol.dispose](); @@ -34,7 +34,7 @@ describe('shared primitives across workers', () => { counter.store(42); const run = workers(2); try { - const [a, b] = await Promise.all([run(rwlockRead(rw, counter)), run(rwlockRead(rw, counter))]); + const [a, b] = await run([rwlockRead(rw, counter), rwlockRead(rw, counter)]); assert.equal(a, 42); assert.equal(b, 42); } finally {