albert/mrfrys-node-server
1
0
Fork 0
mirror of https://gitlab.com/MrFry/mrfrys-node-server synced 2025-04-01 20:24:18 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

restart workers on error

Browse source
This commit is contained in:
mrfry 2022-12-12 20:41:51 +01:00
parent 0259cfe1a7
commit 16cd4d9d4b
4 changed files with 57 additions and 14 deletions
Download patch file Download diff file Expand all files Collapse all files

54
src/utils/workerPool.ts
View file

@ -28,6 +28,8 @@ import { Result, Edits } from './actions'
import type { Question, QuestionDb, QuestionData } from '../types/basicTypes'
import type { WorkerResult } from './classes'
const threadCount = process.env.NS_THREAD_COUNT || os.cpus().length
interface WorkerObj {
worker: Worker
index: number
@ -91,6 +93,7 @@ interface DoneEvent extends EventEmitter {
const alertOnPendingCount = 50
const workerFile = './src/utils/classes.ts'
let workers: Array<WorkerObj>
let getInitData: () => Array<QuestionDb> = null
const pendingJobs: {
[id: string]: PendingJob
} = {}
@ -158,14 +161,16 @@ export function doALongTask(
})
}
export function initWorkerPool(initData: Array<QuestionDb>): Array<WorkerObj> {
export function initWorkerPool(
initDataGetter: () => Array<QuestionDb>
): Array<WorkerObj> {
getInitData = initDataGetter
if (workers) {
logger.Log('WORKERS ALREADY EXISTS', logger.GetColor('redbg'))
return null
}
workers = []
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_WORKER_COUNT: '${threadCount}'`,
@ -175,7 +180,7 @@ export function initWorkerPool(initData: Array<QuestionDb>): Array<WorkerObj> {
for (let i = 0; i < threadCount; i++) {
workers.push({
worker: getAWorker(i, initData),
worker: getAWorker(i, getInitData()),
index: i,
free: true,
})
@ -239,10 +244,10 @@ function processJob() {
}
}
function getAWorker(i: number, initData: Array<QuestionDb>) {
function getAWorker(workerIndex: number, initData: Array<QuestionDb>) {
const worker = workerTs(workerFile, {
workerData: {
workerIndex: i,
workerIndex: workerIndex,
initData: initData,
},
})
@ -254,16 +259,43 @@ function getAWorker(i: number, initData: Array<QuestionDb>) {
console.error(err)
})
worker.on('exit', (code) => {
// TODO: this is critical, whole server should stop, or child threads should be restarted
logger.Log(
`[MAIN]: worker #${i} exit code: ${code}`,
code === 0 ? logger.GetColor('redbg') : logger.GetColor('green')
)
worker.on('exit', (exitCode) => {
handleWorkerExit(workerIndex, exitCode)
})
return worker
}
function handleWorkerExit(exitedWorkerIndex: number, exitCode: number) {
logger.Log(
`[THREAD #${exitedWorkerIndex}]: exit code: ${exitCode}`,
logger.GetColor('redbg')
)
const exitedWorker = workers.find((worker) => {
return worker.index === exitedWorkerIndex
})
try {
exitedWorker.worker.removeAllListeners()
exitedWorker.worker.terminate()
} catch (e) {
console.log(e)
}
workers = workers.filter((worker) => {
return worker.index !== exitedWorkerIndex
})
if (workers.length < threadCount) {
logger.Log(`[THREAD #${exitedWorkerIndex}]: Restarting ... `)
workers.push({
worker: getAWorker(exitedWorkerIndex, getInitData()),
index: exitedWorkerIndex,
free: true,
})
}
}
// ---------------------------------------------------------------------------
function doSomething(currWorker: WorkerObj, obj: TaskObject) {