feat: implement pool() for controlled concurrency and add tests #1375

src/core.ts

@@ -1085,3 +1085,27 @@ export function resolveDefaults(
     return m
   }, defs)
 }
+
+/**
+ * Runs tasks with limited concurrency.
+ * @param limit Max number of concurrent tasks.
+ * @param tasks Array of functions returning promises.
+ */
+export async function pool<T>(
+  limit: number,
+  tasks: (() => Promise<T>)[]
+): Promise<T[]> {
+  const results = new Array<T>(tasks.length)
+  const queue = tasks.entries()
+
+  const worker = async () => {
+    for (const [index, task] of queue) {
+      results[index] = await task()
+    }
+  }
+
+  const workers = Array.from({ length: Math.min(limit, tasks.length) }, worker)
+  await Promise.all(workers)
+
+  return results
+}

test/pool.test.js

@@ -0,0 +1,27 @@
+import { test } from 'node:test'
+import assert from 'node:assert'
+import { $, pool, within, quote } from '../src/index.ts'
+
+test('pool() limits concurrency', async () => {
+  await within(async () => {
+    $.quote = quote
+    $.shell = true
+
+    console.log(' Running concurrency test...')
+    const start = Date.now()
+    const duration = 1000
+
+    await pool(2, [
+      () => $`node -e "setTimeout(() => {}, ${duration})"`,
+      () => $`node -e "setTimeout(() => {}, ${duration})"`,
+      () => $`node -e "setTimeout(() => {}, ${duration})"`,
+      () => $`node -e "setTimeout(() => {}, ${duration})"`,
+    ])
+
+    const totalDuration = Date.now() - start
+    console.log(`Total time: ${totalDuration}ms`)
+
+    assert.ok(totalDuration >= 2000, `Too fast! Got ${totalDuration}ms`)
+    assert.ok(totalDuration < 3500, `Too slow! Got ${totalDuration}ms`)
+  })
+})