Press n or j to go to the next uncovered block, b, p or k for the previous block.
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 | 2x 2x 9x 9x 9x 9x 1x 51x 51x 51x 29x 22x 51x 51x 51x 51x 51x 53x 53x 22x 22x 22x 51x 49x 49x 51x 2x 2x 2x 51x 51x 2x 2x | import { Worker } from 'worker_threads';
import path from 'path';
import { fileURLToPath } from 'url';
const __filename = fileURLToPath(import.meta.url);
const __dirname = path.dirname(__filename);
/**
* Runs background jobs in worker threads for CPU-intensive operations
* without blocking the main thread. Implements a worker pool to limit
* concurrent workers and prevent memory issues.
*/
export class BackgroundJobRunner {
maxConcurrency;
activeWorkers = 0;
maxActiveWorkers = 0; // Track peak concurrency for testing
// Using `any` here because this queue holds jobs with different return types.
// The generic type T varies per job, and TypeScript's variance rules prevent
// using `unknown` with the resolve callback. Type safety is enforced at the
// API boundary via the generic run<T>() method.
// eslint-disable-next-line @typescript-eslint/no-explicit-any
jobQueue = [];
constructor(maxConcurrency = 4) {
this.maxConcurrency = maxConcurrency;
}
/**
* Get the current number of active workers (for testing)
*/
getActiveWorkers() {
return this.activeWorkers;
}
/**
* Get the maximum number of concurrent workers reached (for testing)
*/
getMaxActiveWorkers() {
return this.maxActiveWorkers;
}
/**
* Execute a job in a worker thread.
* @param action - Path to the job module
* @param params - Parameters to pass to the job
* @returns Promise that resolves with the job result
*/
run(action, params) {
return new Promise((resolve, reject) => {
const job = { action, params, resolve, reject };
// If we're under the concurrency limit, start the job immediately
if (this.activeWorkers < this.maxConcurrency) {
this.executeJob(job);
}
else {
// Otherwise, add to queue
this.jobQueue.push(job);
}
});
}
/**
* Execute a queued job in a worker thread
*/
executeJob(job) {
this.activeWorkers++;
this.maxActiveWorkers = Math.max(this.maxActiveWorkers, this.activeWorkers);
// console.log(`Starting background job (${this.activeWorkers}/${this.maxConcurrency} active):`, job.action)
const workerWrapperPath = path.join(__dirname, 'worker-wrapper.js');
// Create a worker thread to run the job action
const worker = new Worker(workerWrapperPath, {
workerData: {
actionPath: job.action,
params: job.params
}
});
const cleanup = () => {
this.activeWorkers--;
// Start next job from queue if available
if (this.jobQueue.length > 0) {
const nextJob = this.jobQueue.shift();
Eif (nextJob) {
this.executeJob(nextJob);
}
}
};
worker.on('message', (result) => {
cleanup();
job.resolve(result);
});
worker.on('error', (error) => {
console.log('Background job error:', error.message || error);
cleanup();
job.reject(error);
});
worker.on('exit', (code) => {
if (code !== 0) {
// console.log(`Background job exited with code ${code}`)
cleanup();
job.reject(new Error(`Worker stopped with exit code ${code}`));
}
});
}
}
|