import type * as d from '../../declarations' ; import { EventEmitter } from 'events' ; import { TASK_CANCELED_MSG } from '@utils' ; import { NodeWorkerMain } from './node-worker-main' ; import { cpus } from 'os' ; export class NodeWorkerController extends EventEmitter implements d . WorkerMainController { workerIds = 0 ; stencilId = 0 ; isEnding = false ; taskQueue : d . CompilerWorkerTask [ ] = [ ] ; workers : NodeWorkerMain [ ] = [ ] ; maxWorkers : number ; useForkedWorkers : boolean ; mainThreadRunner : { [ fnName ] : ( ... args : any [ ] ) => Promise < any > ; } ; constructor ( public forkModulePath , maxConcurrentWorkers ) { super ( ) ; const osCpus = cpus ( ) . length ; this . useForkedWorkers = maxConcurrentWorkers > 0 ; this . maxWorkers = Math . max ( Math . min ( maxConcurrentWorkers , osCpus ) , 2 ) - 1 ; if ( this . useForkedWorkers ) { this . startWorkers ( ) ; } else { this . mainThreadRunner = require ( forkModulePath ) ; } } onError ( err , workerId ) { if ( err . code === 'ERR_IPC_CHANNEL_CLOSED' ) { return this . stopWorker ( workerId ) ; } if ( err . code !== 'EPIPE' ) { console . error ( err ) ; } } onExit ( workerId ) { setTimeout ( ( ) => { let doQueue = false ; const worker = this . workers . find ( ( w ) => w . id === workerId ) ; if ( worker ) { worker . tasks . forEach ( ( t ) => { t . retries ++ ; this . taskQueue . unshift ( t ) ; doQueue = true ; } ) ; worker . tasks . clear ( ) ; } this . stopWorker ( workerId ) ; if ( doQueue ) { this . processTaskQueue ( ) ; } } , 10 ) ; } startWorkers ( ) { while ( this . workers . length < this . maxWorkers ) { this . startWorker ( ) ; } } startWorker ( ) { const workerId = this . workerIds ++ ; const worker = new NodeWorkerMain ( workerId , this . forkModulePath ) ; worker . on ( 'response' , this . processTaskQueue . bind ( this ) ) ; worker . once ( 'exit' , ( ) => { this . onExit ( workerId ) ; } ) ; worker . on ( 'error' , ( err ) => { this . onError ( err , workerId ) ; } ) ; this . workers . push ( worker ) ; } stopWorker ( workerId ) { const worker = this . workers . find ( ( w ) => w . id === workerId ) ; if ( worker ) { worker . stop ( ) ; const index = this . workers . indexOf ( worker ) ; if ( index > - 1 ) { this . workers . splice ( index , 1 ) ; } } } processTaskQueue ( ) { if ( this . isEnding ) { return ; } if ( this . useForkedWorkers ) { this . startWorkers ( ) ; } while ( this . taskQueue . length > 0 ) { const worker = getNextWorker ( this . workers ) ; if ( ! worker ) { break ; } worker . run ( this . taskQueue . shift ( ) ) ; } } send ( ... args ) { if ( this . isEnding ) { return Promise . reject ( TASK_CANCELED_MSG ) ; } if ( this . useForkedWorkers ) { return new Promise < any > ( ( resolve , reject ) => { const task = { stencilId : this . stencilId ++ , inputArgs : args , retries : 0 , resolve : resolve , reject : reject } ; this . taskQueue . push ( task ) ; this . processTaskQueue ( ) ; } ) ; } return this . mainThreadRunner [ args [ 0 ] ] . apply ( null , args . slice ( 1 ) ) ; } handler ( name ) { return ( ... args ) => { return this . send ( name , ... args ) ; } ; } cancelTasks ( ) { for ( const worker of this . workers ) { worker . tasks . forEach ( ( t ) => t . reject ( TASK_CANCELED_MSG ) ) ; worker . tasks . clear ( ) ; } this . taskQueue . length = 0 ; } destroy ( ) { if ( ! this . isEnding ) { this . isEnding = true ; for ( const task of this . taskQueue ) { task . reject ( TASK_CANCELED_MSG ) ; } this . taskQueue . length = 0 ; const workerIds = this . workers . map ( ( w ) => w . id ) ; for ( const workerId of workerIds ) { this . stopWorker ( workerId ) ; } } } } export function getNextWorker ( workers ) { const availableWorkers = workers . filter ( ( w ) => { if ( w . stopped ) { return false ; } return true ; } ) ; if ( availableWorkers . length === 0 ) { return null ; } const sorted = availableWorkers . sort ( ( a , b ) => { if ( a . tasks . size < b . tasks . size ) return - 1 ; if ( a . tasks . size > b . tasks . size ) return 1 ; if ( a . totalTasksAssigned < b . totalTasksAssigned ) return - 1 ; if ( a . totalTasksAssigned > b . totalTasksAssigned ) return 1 ; return 0 ; } ) ; return sorted [ 0 ] ; }