import { Worker } from 'worker_threads' ; import path from 'path' ; import { Pool } from 'tarn' ; import { UpdateChannelParams } from '@statechannels/client-api-schema' ; import { Either } from 'fp-ts/lib/Either' ; import { isLeft , isRight } from 'fp-ts/lib/These' ; import _ from 'lodash' ; import P , { Logger } from 'pino' ; import { EngineConfig , MultipleChannelOutput , SingleChannelOutput } from '../types' ; import { StateChannelWorkerData } from './worker-data' ; const ONE_DAY = 86400000 ; export class WorkerManager { private pool ? : Pool < Worker > ; private threadAmount : number ; private logger : Logger ; param engineConfig param onNewWorker constructor ( engineConfig , logger ) { this . logger = logger . child ( { module : 'Worker-Manager' } ) ; this . threadAmount = engineConfig . workerThreadAmount ; if ( this . threadAmount === 0 ) { throw new Error ( 'Invalid engineConfig: threadAmount should not be 0' ) ; } this . pool = new Pool ( { create : ( ) : Worker => { this . logger . trace ( 'Starting worker' ) ; const worker = new Worker ( path . resolve ( __dirname , './loader.js' ) , { workerData : { engineConfig , parentLogger : logger } , } ) ; worker . on ( 'error' , err => { throw err ; } ) ; this . logger . trace ( 'Started worker %o' , worker . threadId ) ; return worker ; } , destroy : ( worker ) : Promise < number > => worker . terminate ( ) , min : this . threadAmount , max : this . threadAmount , reapIntervalMillis : ONE_DAY , idleTimeoutMillis : ONE_DAY , } ) ; } public async warmUpThreads ( ) { this . logger . trace ( 'Warming up threads' ) ; const acquire = _ . range ( this . threadAmount ) . map ( ( ) => this . pool ?. acquire ( ) . promise ) ; const workers = await Promise . all ( acquire ) ; workers . forEach ( w => { if ( w ) this . pool ?. release ( w ) ; else throw Error ( 'No worker acquired' ) ; } ) ; } param { operation returns private async sendOperation < T extends SingleChannelOutput | MultipleChannelOutput , O extends StateChannelWorkerData > ( { operation , args } ) { this . logger . trace ( '%s called' , operation ) ; if ( ! this . pool ) throw new Error ( `Worker threads are disabled` ) ; const worker = await this . pool . acquire ( ) . promise ; const resultPromise = new Promise < T > ( ( resolve , reject ) => { const handler = ( response ) => { this . pool ?. release ( worker ) ; if ( isLeft ( response ) ) { reject ( response . left ) ; } else if ( isRight ( response ) ) { worker . off ( 'message' , handler ) ; resolve ( response . right ) ; } } ; worker . on ( 'message' , handler ) ; } ) ; worker . postMessage ( { operation , args } ) ; return resultPromise ; } public async pushMessage ( args ) { return this . sendOperation ( { operation : 'PushMessage' , args } ) ; } public async pushUpdate ( args ) { return this . sendOperation ( { operation : 'PushUpdate' , args } ) ; } public async updateChannel ( args ) { return this . sendOperation ( { operation : 'UpdateChannel' , args } ) ; } public async destroy ( ) { await this . pool ?. destroy ( ) ; } }