Repository: marpple/FxTS (ts)

Issue:
Q. Is there a good way to implement concurrency using a queue-like approach instead of the `concurrent` function?
Q. When implementing concurrency for asynchronous functions, the documentation recommends using `concurrent`.

But, the problem with both `concurrent` and `Promise.all` is that even if some functions finish quickly, the overall execution time is still determined by the slowest asynchronous function.

I’d like to implement, once some asynchronous functions finish early, the remaining pending functions are immediately queued—ensuring that the desired concurrency level is consistently maintained throughout the entire loop over the collection.

To achieve this, I am currently using an external library like [p-queue](https://github.com/sindresorhus/p-queue), as shown in the example below.

```ts
import PQueue from "p-queue"

const concurrency = 3
const q = new PQueue({ concurrency })
const now = performance.now()

pipe(
  range(Number.MAX_SAFE_INTEGER),
  take(99),
  map((i) => async () => {
    await delay(i * 100)
    console.log(i, Math.floor(performance.now() - now) + "ms")
    return i
  }),
  map(q.add.bind(q)),
  toArray,
)

// 0 3ms
// 1 102ms
// 2 202ms
// 3 305ms
// 4 503ms
// 5 703ms
// 6 905ms
// 7 1204ms
// 8 1504ms
// 9 1806ms
// 10 2205ms
// 11 2605ms
// 12 3007ms
// ...
```

compare to:

```ts
await pipe(
  range(Number.MAX_SAFE_INTEGER),
  toAsync,
  map(async (i) => {
    await delay(i * 100)
    console.log(i, Math.floor(performance.now() - now) + "ms")
    return i
  }),
  take(999),
  concurrent(3),
  toArray,
)

// 0 1ms
// 1 101ms
// 2 201ms
// 3 503ms
// 4 604ms
// 5 703ms
// 6 1305ms
// 7 1405ms
// 8 1505ms
// 9 2406ms
// 10 2506ms
// 11 2607ms
// 12 3809ms
// ...
```

Is there a native way in `@fxts/core` to accomplish this without relying on external libraries like `p-queue`?

---

Q. The `range` function accepts `Infinity` as an argument, but why can’t the `concurrent` function accept `Infinity` (like same behavior with `Promise.all`)?
Is this an intentional design decision, or is there room for improvement?

Pull request:
feat: add `concurrentPool`
Fixes #304

I'd like to get feedback from you  @ysm-dev 
What do you think about the expression and behavior?


Patch:
diff --git a/src/Lazy/concurrentPool.ts b/src/Lazy/concurrentPool.ts
new file mode 100644
index 00000000..f4020970
--- /dev/null
+++ b/src/Lazy/concurrentPool.ts
@@ -0,0 +1,182 @@
+import { isAsyncIterable } from "../_internal/utils";
+import identity from "../identity";
+import type { Reject, Resolve } from "../types/Utils";
+import map from "./map";
+
+type Item<T> = {
+  fail?: unknown;
+  success:
+    | { value: undefined; done: true } // done
+    | { value: T; done: false };
+};
+
+/**
+ * Ensures consistency in the level of concurrency based on the given number of concurrent tasks.
+ *
+ * @example
+ * ```ts
+ * await pipe(
+ *   [1, 2, 3, 4, 5, 6],
+ *   toAsync,
+ *   map((a) => delay(100 * a, a)),
+ *   concurrentPool(3),
+ *   each(console.log), // log 1, 2, 3, 4, 5, 6
+ * ); // 2 seconds
+ */
+function concurrentPool<A>(
+  length: number,
+  iterable: AsyncIterable<A>,
+): AsyncIterableIterator<A>;
+
+function concurrentPool<A>(
+  length: number,
+  iterable?: AsyncIterable<A>,
+): (iterable: AsyncIterable<A>) => AsyncIterableIterator<A>;
+
+function concurrentPool<A>(
+  length: number,
+  iterable?: AsyncIterable<A>,
+):
+  | AsyncIterableIterator<A>
+  | ((iterable: AsyncIterable<A>) => AsyncIterableIterator<A>) {
+  if (iterable === undefined) {
+    return (iterable) => {
+      return concurrentPool(length, iterable);
+    };
+  }
+
+  if (!Number.isFinite(length) || length <= 0) {
+    throw new RangeError("'length' must be positive integer");
+  }
+  if (!isAsyncIterable(iterable)) {
+    throw new TypeError("'iterable' must be type of AsyncIterable");
+  }
+
+  if (length === 1) {
+    return map(identity, iterable);
+  }
+
+  let fillId = 1;
+  let consumeId = 1;
+  let idleWorkers = length;
+
+  const consumer: [Resolve<A>, Reject, number][] = [];
+  const itemMap: Map<number, Item<A>> = new Map();
+  const iterator = iterable[Symbol.asyncIterator]();
+
+  let finished = false;
+  let prev = Promise.resolve();
+
+  async function fetchNextItem() {
+    if (idleWorkers <= 0) {
+      return;
+    }
+
+    idleWorkers--;
+    const id = fillId++;
+    fillId = fillId % Number.MAX_SAFE_INTEGER;
+    try {
+      const item = {
+        success: await iterator.next(),
+      } as Item<A>;
+      idleWorkers++;
+
+      itemMap.set(id, item);
+      chainItem(item);
+    } catch (error) {
+      idleWorkers++;
+      finished = true;
+
+      const item = {
+        success: { done: false },
+        fail: error,
+      } as Item<A>;
+
+      itemMap.set(id, item);
+      chainItem(item);
+    }
+  }
+
+  function chainItem(item: Item<A>) {
+    prev = prev.then(() => item).then(pullItem);
+    return item;
+  }
+
+  function doneQueue() {
+    while (consumer.length > 0) {
+      // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+      const [resolve] = consumer.shift()!;
+      resolve({
+        value: undefined,
+        done: true,
+      });
+    }
+  }
+
+  async function pullItem(item: Item<A>): Promise<void> {
+    while (consumer.length > 0) {
+      const id = consumer[0][2];
+      if (!itemMap.has(id)) {
+        break;
+      }
+      // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+      const [resolve, reject] = consumer.shift()!;
+      // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+      const value = itemMap.get(id)!;
+      itemMap.delete(id);
+      if (value.fail) {
+        reject(value.fail);
+        doneQueue();
+      } else {
+        resolve(value.success);
+      }
+    }
+
+    /**
+     * @TODO ppeeou: If an error occurs, the next iterator should not be executed
+     */
+    if (!item.success.done && consumer.length > 0) {
+      fetchNextItem();
+    }
+  }
+
+  async function processQueue() {
+    if (idleWorkers == 0) {
+      return;
+    }
+
+    return Promise.all(
+      Array.from({ length: idleWorkers }, () => fetchNextItem()),
+    );
+  }
+
+  function slide() {
+    if (finished && fillId === consumeId) {
+      while (consumer.length > 0) {
+        // eslint-disable-next-line @typescript-eslint/no-non-null-assertion
+        const [resolve] = consumer.shift()!;
+        resolve({ value: undefined, done: true });
+      }
+    } else {
+      processQueue();
+    }
+  }
+
+  return {
+    [Symbol.asyncIterator]() {
+      return this;
+    },
+    async next() {
+      return new Promise((resolve, reject) => {
+        const id = consumeId++;
+        consumeId = consumeId % Number.MAX_SAFE_INTEGER;
+        const task: [Resolve<A>, Reject, number] = [resolve, reject, id];
+
+        consumer.push(task);
+        slide();
+      });
+    },
+  } as AsyncIterableIterator<A>;
+}
+
+export default concurrentPool;
diff --git a/src/Lazy/index.ts b/src/Lazy/index.ts
index 9b876658..e7a90f8a 100644
--- a/src/Lazy/index.ts
+++ b/src/Lazy/index.ts
@@ -4,6 +4,7 @@ import compact from "./compact";
 import compress from "./compress";
 import concat from "./concat";
 import concurrent from "./concurrent";
+import concurrentPool from "./concurrentPool";
 import cycle from "./cycle";
 import difference from "./difference";
 import differenceBy from "./differenceBy";
@@ -50,6 +51,7 @@ export {
   compress,
   concat,
   concurrent,
+  concurrentPool,
   cycle,
   difference,
   differenceBy,
diff --git a/website/function.json b/website/function.json
index 902836aa..c15e68c9 100644
--- a/website/function.json
+++ b/website/function.json
@@ -6,6 +6,7 @@
     "compress",
     "concat",
     "concurrent",
+    "concurrentPool",
     "cycle",
     "drop",
     "dropRight",
diff --git a/website/markdown-plugin/src/function.json b/website/markdown-plugin/src/function.json
index 902836aa..c15e68c9 100644
--- a/website/markdown-plugin/src/function.json
+++ b/website/markdown-plugin/src/function.json
@@ -6,6 +6,7 @@
     "compress",
     "concat",
     "concurrent",
+    "concurrentPool",
     "cycle",
     "drop",
     "dropRight",


Test patch:
diff --git a/test/Lazy/concurrent.spec.ts b/test/Lazy/concurrent.spec.ts
index 65ad9887..7ac0bfa3 100644
--- a/test/Lazy/concurrent.spec.ts
+++ b/test/Lazy/concurrent.spec.ts
@@ -121,6 +121,7 @@ describe("concurrent", function () {
   }, 1050);
 
   it("should be able to handle an error when working concurrent", async function () {
+    const fn = jest.fn();
     const res = concurrent(
       2,
       toAsync(
@@ -142,8 +143,10 @@ describe("concurrent", function () {
         acc.push(item);
       }
     } catch (err) {
+      fn();
       expect(err).toEqual("err");
     }
+    expect(fn).toBeCalled();
     expect(acc).toEqual([1, 2, 3]);
   }, 2050);
 });
diff --git a/test/Lazy/concurrentPool.spec.ts b/test/Lazy/concurrentPool.spec.ts
new file mode 100644
index 00000000..2bb4da91
--- /dev/null
+++ b/test/Lazy/concurrentPool.spec.ts
@@ -0,0 +1,117 @@
+import { concurrentPool, delay, pipe, toAsync } from "../../src/index";
+
+describe("concurrentPool", function () {
+  it("should be consumed 'AsyncIterable' concurrently (concurrency 1)", async function () {
+    const res = concurrentPool(
+      1,
+      toAsync(
+        (function* () {
+          yield delay(1000, 1);
+          yield delay(1000, 2);
+          yield delay(1000, 3);
+        })(),
+      ),
+    );
+    const acc = [];
+    for await (const item of res) {
+      acc.push(item);
+    }
+    expect(acc).toEqual([1, 2, 3]);
+  }, 3050);
+
+  it("should be consumed 'AsyncIterable' concurrently (concurrency 2)", async function () {
+    const res = concurrentPool(
+      2,
+      toAsync(
+        (function* () {
+          yield delay(300, 1);
+          yield delay(1000, 2);
+          yield delay(600, 3);
+          yield delay(1000, 4);
+          yield delay(600, 5);
+          yield delay(300, 6);
+        })(),
+      ),
+    );
+    const acc = [];
+    for await (const item of res) {
+      acc.push(item);
+    }
+    expect(acc).toEqual([1, 2, 3, 4, 5, 6]);
+  }, 1950);
+
+  it("should be consumed 'AsyncIterable' concurrently (concurrency n)", async function () {
+    const res = concurrentPool(
+      3,
+      toAsync(
+        (function* () {
+          yield delay(300, 1);
+          yield delay(1000, 2);
+          yield delay(600, 3);
+          yield delay(1000, 4);
+          yield delay(600, 5);
+          yield delay(300, 6);
+        })(),
+      ),
+    );
+    const acc = [];
+    for await (const item of res) {
+      acc.push(item);
+    }
+    expect(acc).toEqual([1, 2, 3, 4, 5, 6]);
+  }, 1350);
+
+  it("should be able to be used as a curried function in the pipeline", async function () {
+    const iter = pipe(
+      toAsync(
+        (function* () {
+          yield delay(300, 1);
+          yield delay(1000, 2);
+          yield delay(300, 3);
+          yield delay(1000, 4);
+        })(),
+      ),
+      concurrentPool(2),
+    );
+
+    const acc = await Promise.all([
+      iter.next(),
+      iter.next(),
+      iter.next(),
+      iter.next(),
+    ]).then((arr) => arr.map((a) => a.value));
+
+    expect(acc).toEqual([1, 2, 3, 4]);
+  }, 1650);
+
+  it("should be able to handle an error when working concurrentPool", async function () {
+    const fn = jest.fn();
+    const res = concurrentPool(
+      2,
+      toAsync(
+        (function* () {
+          yield delay(100, 1);
+          yield delay(1000, 2);
+          yield delay(100, 3);
+          yield (async () => {
+            throw "err";
+          })(),
+            yield delay(100, 4);
+          yield delay(1000, 5);
+        })(),
+      ),
+    );
+
+    const acc = [];
+    try {
+      for await (const item of res) {
+        acc.push(item);
+      }
+    } catch (err) {
+      fn();
+      expect(err).toEqual("err");
+    }
+    expect(fn).toBeCalled();
+    expect(acc).toEqual([1, 2, 3]);
+  }, 2050);
+});
diff --git a/type-check/Lazy/concurrentPool.test.ts b/type-check/Lazy/concurrentPool.test.ts
new file mode 100644
index 00000000..dddad805
--- /dev/null
+++ b/type-check/Lazy/concurrentPool.test.ts
@@ -0,0 +1,12 @@
+import { concurrentPool, pipe, toAsync } from "../../src";
+import * as Test from "../../src/types/Test";
+
+const { checks, check } = Test;
+
+const res1 = concurrentPool(3, toAsync([1, 2, 3, 4]));
+const res2 = pipe(toAsync([1, 2, 3, 4]), concurrentPool(3));
+
+checks([
+  check<typeof res1, AsyncIterableIterator<number>, Test.Pass>(),
+  check<typeof res2, AsyncIterableIterator<number>, Test.Pass>(),
+]);