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 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 | 2x 2x 2x 2x 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)
interface QueuedJob<T> {
action: string
params?: unknown
resolve: (value: T) => void
reject: (error: Error) => void
}
/**
* 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 {
private maxConcurrency: number
private activeWorkers: number = 0
private maxActiveWorkers: number = 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
private jobQueue: QueuedJob<any>[] = []
constructor(maxConcurrency: number = 4) {
this.maxConcurrency = maxConcurrency
}
/**
* Get the current number of active workers (for testing)
*/
getActiveWorkers(): number {
return this.activeWorkers
}
/**
* Get the maximum number of concurrent workers reached (for testing)
*/
getMaxActiveWorkers(): number {
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<T = unknown>(action: string, params?: unknown): Promise<T> {
return new Promise((resolve, reject) => {
const job: QueuedJob<T> = { 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
*/
private executeJob<T>(job: QueuedJob<T>): void {
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()
if (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}`))
}
})
}
}
|