Created runWorker (#288)

* refactor: Renamed function

* breaking: Made runTask private

* feature: Created runWorker

* chore: Updated tests

* chore: Updated docs

* chore: Added changesets

Author: ijlee2

.changeset/chubby-books-argue.md

@@ -0,0 +1,6 @@
+---
+"@codemod-utils/threads": minor
+"docs-app-for-codemod-utils": patch
+---
+
+Made runTask private

.changeset/empty-carrots-invent.md

@@ -0,0 +1,6 @@
+---
+"@codemod-utils/threads": minor
+"docs-app-for-codemod-utils": patch
+---
+
+Created runWorker

docs/src/docs/packages/codemod-utils-threads.md

@@ -55,7 +55,7 @@ function parallelize<T extends unknown[], U>(
 ): Promise<U[]>;
 ```
 
-```ts [Example (src/steps/analyze-files.ts)]
+```ts [Example (src/steps/analyze-files.ts)]{12-15}
 import type { Options } from '../types/index.js';
 import { task } from './analyze-files/task.js';
 
@@ -83,33 +83,19 @@ export function task(filePath: string, projectRoot: string): Result {
 ```
 
 ```ts [Example (src/steps/analyze-files/worker.ts)]
-import { parentPort, workerData } from 'node:worker_threads';
-
-import { runTask } from '@codemod-utils/threads';
+import { runWorker } from '@codemod-utils/threads';
 
 import { task } from './task.js';
 
-type WorkerData = {
-  datasets: Parameters<typeof task>[];
-};
-
-const { datasets } = workerData as WorkerData;
-
-runTask(task, datasets)
-  .then((result) => {
-    parentPort?.postMessage(result);
-  })
-  .catch((error) => {
-    throw error;
-  });
+runWorker(task);
 ```
 
 :::
 
 
-### runTask {#api-run-task}
+### runWorker {#api-run-worker}
 
-Runs a task on many datasets. The size of datasets should be moderate. Primarily used to create a worker file for a task.
+Runs a task on a worker.
 
 > [!TIP]
 > 
@@ -122,43 +108,18 @@ Runs a task on many datasets. The size of datasets should be moderate. Primarily
  * @param task
  *
  * Some function to call.
- *
- * @param datasets
- *
- * An array of dataset's.
- * 
- * @return
- * 
- * An array of the task's return value.
  */
 type Task<T extends unknown[], U> = (...dataset: T) => U | Promise<U>;
 
-function runTask<T extends unknown[], U>(
-  task: Task<T, U>,
-  datasets: T[],
-): Promise<U[]>;
+function runWorker<T extends unknown[], U>(task: Task<T, U>): void;
 ```
 
 ```ts [Example (Worker)]{5}
-import { parentPort, workerData } from 'node:worker_threads';
-
-import { runTask } from '@codemod-utils/threads';
+import { runWorker } from '@codemod-utils/threads';
 
 import { task } from './task.js';
 
-type WorkerData = {
-  datasets: Parameters<typeof task>[];
-};
-
-const { datasets } = workerData as WorkerData;
-
-runTask(task, datasets)
-  .then((result) => {
-    parentPort?.postMessage(result);
-  })
-  .catch((error) => {
-    throw error;
-  });
+runWorker(task);
 ```
 
 :::

packages/threads/src/-private/parallelize.ts

@@ -22,12 +22,12 @@ export function batchDatasets<T>(
   return [datasetsForMainThread, datasetsForWorkerThreads];
 }
 
-export function createRunWorker<U>(
+export function getCreateWorker<U>(
   workerOptions: WorkerOptions,
 ): <T>(datasets: T[]) => Promise<U[]> {
   const { importMetaUrl, workerFilePath } = workerOptions;
 
-  function runWorker<T>(datasets: T[]): Promise<U[]> {
+  function createWorker<T>(datasets: T[]): Promise<U[]> {
     return new Promise((resolve, reject) => {
       const workerUrl = new URL(workerFilePath, importMetaUrl);
 
@@ -48,7 +48,7 @@ export function createRunWorker<U>(
     });
   }
 
-  return runWorker;
+  return createWorker;
 }
 
 export function validateWorkerFilePath(workerOptions: WorkerOptions): void {

packages/threads/src/-private/run-task.ts

@@ -1,4 +1,4 @@
-export class Queue<U> {
+class Queue<U> {
   private error: Error | undefined = undefined;
   declare private maxNumTasksRunning: number;
   private results: U[] = [];
@@ -62,3 +62,34 @@ export class Queue<U> {
     await Promise.race(this.tasksRunning);
   }
 }
+
+const MAX_NUM_TASKS_RUNNING = 10;
+
+export type Task<T extends unknown[], U> = (...dataset: T) => U | Promise<U>;
+
+export async function runTask<T extends unknown[], U>(
+  task: Task<T, U>,
+  datasets: T[],
+): Promise<U[]> {
+  const queue = new Queue<U>(MAX_NUM_TASKS_RUNNING);
+
+  for (const dataset of datasets) {
+    if (queue.hasTooManyTasks) {
+      await queue.wait();
+    }
+
+    await queue.runTask(task(...dataset));
+  }
+
+  if (queue.hasTasks) {
+    await queue.finishTasks();
+  }
+
+  const { output } = queue;
+
+  if (output.status === 'failed') {
+    throw output.error;
+  }
+
+  return output.results;
+}

packages/threads/src/index.ts

@@ -1,2 +1,2 @@
 export * from './parallelize.js';
-export * from './run-task.js';
+export * from './run-worker.js';

packages/threads/src/parallelize.ts

@@ -2,11 +2,11 @@ import { availableParallelism } from 'node:os';
 
 import {
   batchDatasets,
-  createRunWorker,
+  getCreateWorker,
   validateWorkerFilePath,
   type WorkerOptions,
 } from './-private/parallelize.js';
-import { runTask, type Task } from './run-task.js';
+import { runTask, type Task } from './-private/run-task.js';
 
 const MIN_NUM_TASKS_PER_WORKER = 100;
 
@@ -40,6 +40,8 @@ const MIN_NUM_TASKS_PER_WORKER = 100;
  *
  * ```ts
  * // src/steps/analyze-files.ts
+ * import { parallelize } from '@codemod-utils/threads';
+ *
  * import type { Options } from '../types/index.js';
  * import { task } from './analyze-files/task.js';
  *
@@ -69,25 +71,11 @@ const MIN_NUM_TASKS_PER_WORKER = 100;
  *
  * ```ts
  * // src/steps/analyze-files/worker.ts
- * import { parentPort, workerData } from 'node:worker_threads';
- *
- * import { runTask } from '@codemod-utils/threads';
+ * import { runWorker } from '@codemod-utils/threads';
  *
  * import { task } from './task.js';
  *
- * type WorkerData = {
- *   datasets: Parameters<typeof task>[];
- * };
- *
- * const { datasets } = workerData as WorkerData;
- *
- * runTask(task, datasets)
- *   .then((result) => {
- *     parentPort?.postMessage(result);
- *   })
- *   .catch((error) => {
- *     throw error;
- *   });
+ * runWorker(task);
  * ```
  */
 export async function parallelize<T extends unknown[], U>(
@@ -115,11 +103,11 @@ export async function parallelize<T extends unknown[], U>(
     numTasksPerWorker,
   );
 
-  const runWorker = createRunWorker<U>(workerOptions);
+  const createWorker = getCreateWorker<U>(workerOptions);
 
   const [mainThreadResults, ...workerResults] = await Promise.all([
     runTask(task, datasetsForMainThread),
-    ...datasetsForWorkerThreads.map(runWorker),
+    ...datasetsForWorkerThreads.map(createWorker),
   ]);
 
   return [...mainThreadResults, ...workerResults.flat()];

packages/threads/src/run-task.ts

@@ -0,0 +1,39 @@
+import { parentPort, workerData } from 'node:worker_threads';
+
+import { runTask, type Task } from './-private/run-task.js';
+
+/**
+ * Runs a task on a worker.
+ *
+ * Note, a worker file always uses the code shown below. You just need to
+ * import the task from the right file.
+ *
+ * @param task
+ *
+ * Some function to call.
+ *
+ * @example
+ *
+ * Create a worker file for a task.
+ *
+ * ```ts
+ * import { runWorker } from '@codemod-utils/threads';
+ *
+ * import { task } from './task.js';
+ *
+ * runWorker(task);
+ * ```
+ */
+export function runWorker<T extends unknown[], U>(task: Task<T, U>): void {
+  const { datasets } = workerData as {
+    datasets: T[];
+  };
+
+  runTask(task, datasets)
+    .then((result) => {
+      parentPort?.postMessage(result);
+    })
+    .catch((error) => {
+      throw error;
+    });
+}

packages/threads/src/run-worker.ts

@@ -0,0 +1,13 @@
+import { test } from '@codemod-utils/tests';
+
+import { runTask } from '../../../src/-private/run-task.js';
+import { assertOutput, getDatasets, task } from '../../helpers/vector/setup.js';
+
+test('-private | run-task > vector (1)', async function () {
+  const numTasks = 0;
+  const datasets = getDatasets(numTasks);
+
+  const output = await runTask(task, datasets);
+
+  assertOutput(output);
+});