From: Ardis Lu <ardisglu@gmail.com>
Date: Mon, 20 Jan 2025 07:11:23 +0000 (-0800)
Subject: Add maxConcurrent option
X-Git-Tag: v0.1.0~107
X-Git-Url: https://git.ardis.lu/?a=commitdiff_plain;h=2005dd9249b25bb00bd2d216e8c8d5efc2c0d422;p=csvio

Add maxConcurrent option
---

diff --git a/examples/README.md b/examples/README.md
index e9fec20..3d97d78 100644
--- a/examples/README.md
+++ b/examples/README.md
@@ -22,4 +22,6 @@ node ./examples/ex1.js
 
 [Example 6: Handling a flaky transformation (basic retry strategy)](./ex6.js)
 
-[Example 7: Processing rows in batches (multi-row input)](./ex7.js)
\ No newline at end of file
+[Example 7: Processing rows in batches (multi-row input)](./ex7.js)
+
+[Example 8: Processing rows concurrently (async)](./ex8.js)
\ No newline at end of file
diff --git a/examples/data/ex8-in.csv b/examples/data/ex8-in.csv
new file mode 100644
index 0000000..99bd6d1
--- /dev/null
+++ b/examples/data/ex8-in.csv
@@ -0,0 +1,9 @@
+concurrent group
+a
+a
+a
+a
+a
+b
+b
+b
diff --git a/examples/data/ex8-out.csv b/examples/data/ex8-out.csv
new file mode 100644
index 0000000..de20589
--- /dev/null
+++ b/examples/data/ex8-out.csv
@@ -0,0 +1,9 @@
+tick number,concurrent group
+1,a
+1,a
+1,a
+1,a
+1,a
+2,b
+2,b
+2,b
diff --git a/examples/ex8.js b/examples/ex8.js
new file mode 100644
index 0000000..a44f6be
--- /dev/null
+++ b/examples/ex8.js
@@ -0,0 +1,32 @@
+// ## Example 8: Processing rows concurrently (async)
+// 
+// Set the `maxConcurrent` option to a number greater than `1` to process rows in parallel. The transformation function
+// will be automatically turned into a promise if it isn't already async.
+// 
+// Note that execution is blocked until all promises in a concurrent group settle. So if there is one transformation that
+// takes very long, all row processing is blocked on the 1 transformation even if all others in the group have already
+// resolved.
+
+import { setTimeout } from 'node:timers/promises';
+import { CSVReader, CSVTransformer, CSVWriter } from '../src/index.js';
+
+// Returns how many 100ms increments have elapsed since this program began
+const start = performance.now();
+function tick() {
+  const ms = performance.now() - start;
+  return Math.floor(ms / 100);
+}
+
+let firstChunk = true;
+async function parallel(row) {
+  if (firstChunk) {
+    firstChunk = false;
+    return ['tick number', ...row];
+  }
+  await setTimeout(100); // Simulate some slow task, e.g. a network request
+  return [tick(), ...row];
+}
+
+await new CSVReader(new URL('./data/ex8-in.csv', import.meta.url))
+  .pipeThrough(new CSVTransformer(parallel, { includeHeaders: true, maxConcurrent: 5 }))
+  .pipeTo(new CSVWriter(new URL('./data/ex8-out.csv', import.meta.url)));
\ No newline at end of file
diff --git a/src/core.js b/src/core.js
index e3037c8..b3a4c40 100644
--- a/src/core.js
+++ b/src/core.js
@@ -160,6 +160,10 @@ export class CSVReader extends ReadableStream {
  * to the `onError` function. The default value is `null` (errors will not be caught).
  * @property {Number} [maxBatchSize=1] The maximum number of rows that will be passed to the transformation function per function call (greedy).
  * The default value is `1`.
+ * @property {Number} [maxConcurrent=1] The maximum concurrent executions of the transformation function (greedy). The transformation function
+ * will be automatically turned into a promise if it isn't already async. Execution is blocked until all promises in a concurrent group settle
+ * (i.e., if one transformation in a group is hanging, further rows will NOT be processed even if all other transformations in the group are
+ * resolved). The default value is `1`.
  */
 
 /**
@@ -169,6 +173,7 @@ export class CSVReader extends ReadableStream {
 export class CSVTransformer extends TransformStream {
   #fn;
   #options;
+  #concurrent = [];
   #batch = [];
   #firstChunk = true;
 
@@ -201,6 +206,7 @@ export class CSVTransformer extends TransformStream {
     options.rawOutput ??= false;
     options.onError ??= null;
     options.maxBatchSize ??= 1;
+    options.maxConcurrent ??= 1;
 
     this.#fn = fn;
     this.#options = options;
@@ -228,32 +234,58 @@ export class CSVTransformer extends TransformStream {
     }
   }
 
+  async #enqueueConcurrent(controller) {
+    const results = await Promise.allSettled(this.#concurrent);
+    this.#concurrent.length = 0;
+    for (const r of results) {
+      const { status, value, reason } = r;
+      if (status === 'rejected') {
+        if (reason instanceof Error) {
+          throw reason;
+        }
+        else {
+          throw new Error(reason);
+        }
+      }
+      else {
+        this.#enqueueRow(value, controller);
+      }
+    }
+  }
+
   async #transform(chunk, controller) {
-    const { includeHeaders, rawInput, maxBatchSize } = this.#options;
+    const { includeHeaders, rawInput, maxBatchSize, maxConcurrent } = this.#options;
     const row = rawInput ? chunk : JSON.parse(chunk);
-    let out;
+
     if (this.#firstChunk) {
       this.#firstChunk = false;
-      out = includeHeaders ? await this.#wrappedFn(row) : row;
+      const out = includeHeaders ? await this.#wrappedFn(row) : row;
+      this.#enqueueRow(out, controller);
+      return;
     }
-    else if (maxBatchSize > 1) {
+
+    if (maxBatchSize > 1) {
       this.#batch.push(row);
       if (this.#batch.length === maxBatchSize) {
-        out = await this.#wrappedFn(this.#batch);
+        this.#concurrent.push(this.#wrappedFn(this.#batch));
         this.#batch.length = 0;
       }
     }
     else {
-      out = await this.#wrappedFn(row);
+      this.#concurrent.push(this.#wrappedFn(row));
     }
 
-    this.#enqueueRow(out, controller);
+    if (this.#concurrent.length === maxConcurrent) {
+      await this.#enqueueConcurrent(controller);
+    }
   }
 
   async #flush(controller) {
     if (this.#batch.length > 0) {
-      const out = await this.#wrappedFn(this.#batch);
-      this.#enqueueRow(out, controller);
+      this.#concurrent.push(this.#wrappedFn(this.#batch));
+    }
+    if (this.#concurrent.length > 0) {
+      await this.#enqueueConcurrent(controller);
     }
   }
 }