Commit 6a785867 authored by Cédric Moreau's avatar Cédric Moreau
Browse files

[fix] tests: remove EventEmitter leaks

parent 4aa66888

app/modules/prover/lib/blockProver.ts

+1 −1
Original line number Diff line number Diff line
@@ -73,7 +73,7 @@ export class WorkerFarm { 
  }



  shutDownEngine() {

    this.theEngine.shutDown()

    return this.theEngine.shutDown()

  }



  /**

app/modules/prover/lib/permanentProver.ts

+2 −0
Original line number Diff line number Diff line
@@ -213,6 +213,8 @@ export class PermanentProver { 
    await this.prover.cancel();

    // If we were waiting, stop it and process the continuous generation

    this.blockchainChangedResolver && this.blockchainChangedResolver();

    const farm = await this.prover.getWorker()

    await farm.shutDownEngine()

  }



  private checkTrialIsNotTooHigh(trial:number, current:DBBlock, selfPubkey:string) {

app/modules/prover/lib/powCluster.ts

+38 −24
Original line number Diff line number Diff line
@@ -3,12 +3,11 @@ import {ProverConstants} from "./constants" 


const _ = require('underscore')

const nuuid = require('node-uuid');

const moment = require('moment');

const cluster = require('cluster')

const querablep = require('querablep')

const logger = require('../../../lib/logger').NewLogger()



let clusterId = 0

cluster.setMaxListeners(3)



/**

 * Cluster controller, handles the messages between the main program and the PoW cluster.
@@ -25,6 +24,9 @@ export class Master { 
  logger:any

  onInfoCallback:any

  workersOnline:Promise<any>[]

  private exitHandler: (worker: any, code: any, signal: any) => void

  private onlineHandler: (worker: any) => void

  private messageHandler: (worker: any, msg: any) => void



  constructor(private nbCores:number, logger:any) {

    this.clusterId = clusterId++
@@ -32,6 +34,29 @@ export class Master { 
    this.onInfoMessage = (message:any) => {

      this.logger.info(`${message.pow.pow} nonce = ${message.pow.block.nonce}`)

    }



    this.exitHandler = (worker:any, code:any, signal:any) => {

      this.logger.info(`worker ${worker.process.pid} died with code ${code} and signal ${signal}`)

    }



    this.onlineHandler = (worker:any) => {

      // We just listen to the workers of this Master

      if (this.slavesMap[worker.id]) {

        this.logger.info(`[online] worker c#${this.clusterId}#w#${worker.id}`)

        this.slavesMap[worker.id].online.extras.resolve()

        worker.send({

          command: 'conf',

          value: this.conf

        })

      }

    }



    this.messageHandler = (worker:any, msg:any) => {

      // Message for this cluster

      if (this.slavesMap[worker.id]) {

        this.onWorkerMessage(worker, msg)

      }

    }

  }



  get nbWorkers() {
@@ -62,6 +87,10 @@ export class Master { 
    // this.logger.debug(`ENGINE c#${this.clusterId}#${this.slavesMap[worker.id].index}:`, message)

  }



  /*****************

   * CLUSTER METHODS

   ****************/



  initCluster() {

    // Setup master

    cluster.setupMaster({
@@ -93,28 +122,9 @@ export class Master { 
      return this.slavesMap[worker.id]

    })



    cluster.on('exit', (worker:any, code:any, signal:any) => {

      this.logger.info(`worker ${worker.process.pid} died with code ${code} and signal ${signal}`)

    })



    cluster.on('online', (worker:any) => {

      // We just listen to the workers of this Master

      if (this.slavesMap[worker.id]) {

        this.logger.info(`[online] worker c#${this.clusterId}#w#${worker.id}`)

        this.slavesMap[worker.id].online.extras.resolve()

        worker.send({

          command: 'conf',

          value: this.conf

        })

      }

    })



    cluster.on('message', (worker:any, msg:any) => {

      // Message for this cluster

      if (this.slavesMap[worker.id]) {

        this.onWorkerMessage(worker, msg)

      }

    })

    cluster.on('exit', this.exitHandler)

    cluster.on('online', this.onlineHandler)

    cluster.on('message', this.messageHandler)



    this.workersOnline = this.slaves.map((s:any) => s.online)

    return Promise.all(this.workersOnline)
@@ -165,7 +175,11 @@ export class Master { 
      await Promise.all(this.slaves.map(async (s:any) => {

        s.worker.kill()

      }))

      cluster.removeListener('exit', this.exitHandler)

      cluster.removeListener('online', this.onlineHandler)

      cluster.removeListener('message', this.messageHandler)

    }

    this.slaves = []

  }



  proveByWorkers(stuff:any) {

app/modules/prover/lib/proof.ts

+4 −0
Original line number Diff line number Diff line
@@ -28,6 +28,10 @@ process.on('uncaughtException', (err:any) => { 
  }

});



process.on('unhandledRejection', () => {

  process.exit()

})



process.on('message', async (message) => {



  switch (message.command) {

index.ts

+40 −26
Original line number Diff line number Diff line
@@ -28,6 +28,8 @@ const daemonDependency = require('./app/modules/daemon'); 
const pSignalDependency   = require('./app/modules/peersignal');

const pluginDependency    = require('./app/modules/plugin');



let sigintListening = false



class Stacks {



  static todoOnRunDone:() => any = () => process.exit()
@@ -157,6 +159,8 @@ export interface TransformableDuniterService extends DuniterService, stream.Tran 


class Stack {



  private injectedServices = false



  private cli:any

  private configLoadingCallbacks:any[]

  private configBeforeSaveCallbacks:any[]
@@ -279,9 +283,11 @@ class Stack { 
    }



    const server = new Server(home, program.memory === true, commandLineConf(program));

    let piped = false



    // If ever the process gets interrupted

    let isSaving = false;

    if (!sigintListening) {

      process.on('SIGINT', async () => {

        if (!isSaving) {

          isSaving = true;
@@ -294,7 +300,9 @@ class Stack { 
            process.exit(3);

          }

        }

    });

      })

      sigintListening = true

    }



    // Config or Data reset hooks

    server.resetDataHook = async () => {
@@ -366,6 +374,8 @@ class Stack { 
       * Service injection

       * -----------------

       */

      if (!this.injectedServices) {

        this.injectedServices = true

        for (const def of this.definitions) {

          if (def.service) {

            // To feed data coming from some I/O (network, disk, other module, ...)
@@ -386,6 +396,8 @@ class Stack { 
            }

          }

        }

      }

      piped = true

      // All inputs write to global INPUT stream

      for (const module of this.streams.input) module.pipe(this.INPUT);

      // All processes read from global INPUT stream
@@ -408,13 +420,6 @@ class Stack { 
          const modules = this.streams.input.concat(this.streams.process).concat(this.streams.output).concat(this.streams.neutral);

          // Any streaming module must implement a `stopService` method

          await Promise.all(modules.map((module:DuniterService) => module.stopService()))

          // // Stop reading inputs

          // for (const module of streams.input) module.unpipe();

          // Stop reading from global INPUT

          // INPUT.unpipe();

          // for (const module of streams.process) module.unpipe();

          // // Stop reading from global PROCESS

          // PROCESS.unpipe();

        },



        this);
@@ -422,6 +427,15 @@ class Stack { 
    } catch (e) {

      server.disconnect();

      throw e;

    } finally {

      if (piped) {

        // Unpipe everything, as the command is done

        for (const module of this.streams.input) module.unpipe()

        for (const module of this.streams.process) module.unpipe()

        for (const module of this.streams.output) module.unpipe()

        this.INPUT.unpipe()

        this.PROCESS.unpipe()

      }

    }

  }