Skip to content
Open
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
59 changes: 59 additions & 0 deletions src/classes/child-pool.ts
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,21 @@
free: { [key: string]: Child[] } = {};
private opts: ChildPoolOpts;

/**
* Creates a new ChildPool that manages a set of sandboxed child processes
* (or worker threads) used to run job processors.
*
* @param opts - Pool options.
* @param opts.mainFile - Path to the main bootstrap file loaded inside each

Check failure on line 29 in src/classes/child-pool.ts

View workflow job for this annotation

GitHub Actions / smoke (lint + fast tests on node@lts, ioredis, redis@7)

tsdoc-param-tag-with-invalid-name: The @param block should be followed by a valid parameter name: The identifier cannot non-word characters
* child. Defaults to the bundled CJS or ESM `main.js` depending on the
* runtime module system.
* @param opts.useWorkerThreads - If true, spawn worker threads instead of

Check failure on line 32 in src/classes/child-pool.ts

View workflow job for this annotation

GitHub Actions / smoke (lint + fast tests on node@lts, ioredis, redis@7)

tsdoc-param-tag-with-invalid-name: The @param block should be followed by a valid parameter name: The identifier cannot non-word characters
* forked child processes.
* @param opts.workerForkOptions - Options forwarded to `child_process.fork`

Check failure on line 34 in src/classes/child-pool.ts

View workflow job for this annotation

GitHub Actions / smoke (lint + fast tests on node@lts, ioredis, redis@7)

tsdoc-param-tag-with-invalid-name: The @param block should be followed by a valid parameter name: The identifier cannot non-word characters
* when not using worker threads.
* @param opts.workerThreadsOptions - Options forwarded to the `Worker`

Check failure on line 36 in src/classes/child-pool.ts

View workflow job for this annotation

GitHub Actions / smoke (lint + fast tests on node@lts, ioredis, redis@7)

tsdoc-param-tag-with-invalid-name: The @param block should be followed by a valid parameter name: The identifier cannot non-word characters
* constructor when using worker threads.
*/
constructor({
mainFile = supportCJS()
? path.join(process.cwd(), 'dist/cjs/classes/main.js')
Expand All @@ -37,6 +52,14 @@
};
}

/**
* Retains a child for the given processor file. Reuses a free child when
* one is available, otherwise spawns and initializes a new one.
*
* @param processFile - Absolute path to the processor file the child will
* load and execute jobs from.
* @returns A ready-to-use child instance bound to `processFile`.
*/
async retain(processFile: string): Promise<Child> {
let child = this.getFree(processFile).pop();

Expand Down Expand Up @@ -72,11 +95,23 @@
}
}

/**
* Releases a previously retained child back to the free pool so it can be
* reused by a subsequent `retain` call.
*
* @param child - The child instance to release.
*/
release(child: Child): void {
delete this.retained[child.pid];
this.getFree(child.processFile).push(child);
}

/**
* Removes a child from both the retained map and the free pool. Typically
* called when the underlying process or worker has exited.
*
* @param child - The child instance to remove from the pool.
*/
remove(child: Child): void {
delete this.retained[child.pid];

Expand All @@ -88,6 +123,13 @@
}
}

/**
* Removes a child from the pool and terminates it with the given signal.
* If the child does not exit within the kill timeout it is force-killed.
*
* @param child - The child instance to terminate.
* @param signal - Signal used to terminate the child. Defaults to `SIGKILL`.
*/
async kill(
child: Child,
signal: 'SIGTERM' | 'SIGKILL' = 'SIGKILL',
Expand All @@ -96,6 +138,11 @@
return child.kill(signal, CHILD_KILL_TIMEOUT);
}

/**
* Terminates every child currently tracked by the pool (both retained and
* free) and clears all internal references. Resolves once every child has
* been killed.
*/
async clean(): Promise<void> {
const children = Object.values(this.retained).concat(this.getAllFree());
this.retained = {};
Expand All @@ -104,10 +151,22 @@
await Promise.all(children.map(c => this.kill(c, 'SIGTERM')));
}

/**
* Returns the array of free children for a given processor file, creating
* an empty entry on first access.
*
* @param id - Processor file path used as the pool key.
* @returns The mutable array of free children for `id`.
*/
getFree(id: string): Child[] {
return (this.free[id] = this.free[id] || []);
}

/**
* Returns every free child across all processor files in the pool.
*
* @returns A flat array containing every free child instance.
*/
getAllFree(): Child[] {
return Object.values(this.free).reduce(
(first, second) => first.concat(second),
Expand Down
Loading