Merge

src/utils/workerPool.ts

@@ -1,105 +1,173 @@
 import { Worker } from 'worker_threads'
-import genericPool from 'generic-pool'
+import { v4 as uuidv4 } from 'uuid'
+import { EventEmitter } from 'events'
 import os from 'os'
 
 import logger from './logger'
-// import { QuestionDb } from '../types/basicTypes'
 
-// ---------------------------------------------------------------------------
+interface WorkerObj {
+  worker: any
+  index: number
+  free: Boolean
+}
+
+interface PendingJob {
+  workData: any
+  doneEvent: any
+  targetWorkerIndex?: number
+}
 
 const alertOnPendingCount = 10
 const workerFile = './src/utils/classes.ts'
-let pool: any = null
-let workers: any = null
+let workers: Array<WorkerObj>
+const pendingJobs: {
+  [id: string]: PendingJob
+} = {}
+
+const jobEvents = new EventEmitter()
+
+jobEvents.on('jobDone', () => {
+  processJob()
+})
+
+jobEvents.on('newJob', () => {
+  processJob()
+})
 
 // ---------------------------------------------------------------------------
 
-function handleWorkerError(client, err) {
+function handleWorkerError(worker: WorkerObj, err) {
+  // TODO: restart worker if exited or things like that
   logger.Log('resourcePromise error', logger.GetColor('redbg'))
   console.error(err)
-  pool.destroy(client)
 }
 
-export function doALongTask(obj: any): Promise<any> {
-  if (pool.pending > alertOnPendingCount) {
+// TODO: accuire all workers here, and handle errors so they can be removed if threads exit
+export function msgAllWorker(data: any): Promise<any> {
+  console.log('MSGING ALL WORKER')
+  return new Promise((resolve) => {
+    const promises = []
+    workers.forEach((worker) => {
+      promises.push(doALongTask(data, worker.index))
+    })
+    Promise.all(promises).then((res) => {
+      resolve(res)
+    })
+  })
+}
+
+export function doALongTask(
+  obj: any,
+  targetWorkerIndex?: number
+): Promise<any> {
+  if (Object.keys(pendingJobs).length > alertOnPendingCount) {
     logger.Log(
-      `More than ${alertOnPendingCount} callers waiting for free resource! (${pool.pending})`,
+      `More than ${alertOnPendingCount} callers waiting for free resource! (${
+        Object.keys(pendingJobs).length
+      })`,
       logger.GetColor('redbg')
     )
   }
 
+  const jobId = uuidv4()
+  // FIXME: delete doneEvent?
+  const doneEvent = new EventEmitter()
+  pendingJobs[jobId] = {
+    workData: obj,
+    targetWorkerIndex: targetWorkerIndex,
+    doneEvent: doneEvent,
+  }
+  jobEvents.emit('newJob')
   return new Promise((resolve) => {
-    let currClient
-    pool
-      .acquire()
-      .then(function(client) {
-        currClient = client
-        doSomething(client, obj)
-          .then((res) => {
-            resolve(res)
-            // TODO: check if result is really a result, and want to release port
-            pool.release(client)
-            // console.log('[RELEASE]: #' + client.index)
-          })
-          .catch(function(err) {
-            handleWorkerError(currClient, err)
-          })
-      })
-      .catch(function(err) {
-        handleWorkerError(currClient, err)
-      })
+    doneEvent.once('done', (result) => {
+      jobEvents.emit('jobDone')
+      resolve(result)
+    })
   })
 }
 
 export function initWorkerPool(initData: any): void {
-  if (workers && pool) {
-    logger.Log('WORKER AND POOL ALREADY EXISTS', logger.GetColor('redbg'))
+  if (workers) {
+    logger.Log('WORKERS ALREADY EXISTS', logger.GetColor('redbg'))
     return
   }
   workers = []
   const factory = {
-    create: function() {
-      const currInd = workers.length
-      const worker = getAWorker(currInd, initData)
-      workers.push(worker)
-      return {
-        worker: worker,
-        index: currInd,
-      }
+    create: function(index) {
+      return getAWorker(index, initData)
     },
     destroy: function(client) {
-      // console.log('[DESTROY]')
       client.worker.terminate()
-      // console.log('[DESTROYED] #' + client.index)
     },
   }
 
   const threadCount = process.env.NS_THREAD_COUNT || os.cpus().length
   if (process.env.NS_THREAD_COUNT) {
     logger.Log(
-      `Setting thread count from enviroment variable NS_THREAD_COUNT: '${threadCount}'`,
+      `Setting thread count from enviroment variable NS_WORKER_COUNT: '${workerCount}'`,
       logger.GetColor('red')
     )
   }
 
-  const opts = {
-    min: threadCount, // minimum size of the pool
-    max: threadCount, // maximum size of the pool
-    maxWaitingClients: 999,
+  for (let i = 0; i < workerCount; i++) {
+    workers.push({
+      worker: factory.create(i),
+      index: i,
+      free: true,
+    })
   }
-
-  pool = genericPool.createPool(factory, opts)
-}
-
-// TODO: accuire all workers here, and handle errors so they can be removed if threads exit
-export function msgAllWorker(data: any): void {
-  workers.forEach((worker) => {
-    worker.postMessage(data)
-  })
 }
 
 // ---------------------------------------------------------------------------
 
+function processJob() {
+  if (Object.keys(pendingJobs).length > 0) {
+    // FIXME: FIFO OR ANYTHING ELSE (JOB PROCESSING ORDER)
+    const keys = Object.keys(pendingJobs)
+    let jobKey, freeWorker
+    let i = 0
+    while (!freeWorker && i < keys.length) {
+      jobKey = keys[i]
+      if (!isNaN(pendingJobs[jobKey].targetWorkerIndex)) {
+        if (workers[pendingJobs[jobKey].targetWorkerIndex].free) {
+          freeWorker = workers[pendingJobs[jobKey].targetWorkerIndex]
+          console.log(
+            `RESERVING WORKER ${pendingJobs[jobKey].targetWorkerIndex}`
+          )
+        }
+      } else {
+        freeWorker = workers.find((worker) => {
+          return worker.free
+        })
+        if (freeWorker) {
+          console.log(`RESERVING FIRST AVAILABLE WORKER ${freeWorker.index}`)
+        }
+      }
+      i++
+    }
+
+    if (!freeWorker) {
+      console.log('NO FREE WORKER')
+      return
+    }
+
+    if (freeWorker.free) {
+      freeWorker.free = false
+    }
+    const job = pendingJobs[jobKey]
+    delete pendingJobs[jobKey]
+
+    doSomething(freeWorker, job.workData)
+      .then((res) => {
+        freeWorker.free = true
+        job.doneEvent.emit('done', res)
+      })
+      .catch(function(err) {
+        handleWorkerError(freeWorker, err)
+      })
+  }
+}
+
 function getAWorker(i, initData) {
   const worker = workerTs(workerFile, {
     workerData: {
@@ -110,15 +178,6 @@ function getAWorker(i, initData) {
 
   worker.setMaxListeners(50)
 
-  // worker.on('message', (msg) => {
-  //   logger.Log(`[MAIN]: Msg from worker #${i}`)
-  //   logger.Log(msg)
-  // })
-
-  // worker.on('online', () => {
-  //   logger.Log(`[THREAD #${i}]: Worker ${i} online`)
-  // })
-
   worker.on('error', (err) => {
     logger.Log('Worker error!', logger.GetColor('redbg'))
     console.error(err)
@@ -136,10 +195,9 @@ function getAWorker(i, initData) {
 
 // ---------------------------------------------------------------------------
 
-function doSomething(client, obj) {
-  const { /* index, */ worker } = client
+function doSomething(currWorker, obj) {
+  const { /* index, */ worker } = currWorker
   return new Promise((resolve) => {
-    // console.log('[ACCUIRE]: #' + index)
     worker.postMessage(obj)
     worker.once('message', (msg) => {
       resolve(msg)