ddstats-server/libs/utils/multithread.js

import { randomUUID } from 'crypto'
import { SHARE_ENV, Worker } from 'worker_threads'
import initLog from './log.js'

const log = initLog("Worker Setup")

let workerFarm = []
let workerReady = []
let scheduledJobs = []

function scheduler() {
    if (scheduledJobs.length === 0) return

    const readyIndex = workerReady.indexOf(true)
    if (readyIndex === -1) return

    workerReady[readyIndex] = false
    const worker = workerFarm[readyIndex]

    const job = scheduledJobs.shift()

    worker.postMessage({
        type: 'runScript',
        script: job.script,
        data: job.data,
        id: job.jobID
    })

    const messageHandler = message => {
        if (message.id === job.jobID) {
            workerReady[readyIndex] = true
            worker.removeListener(
                'message',
                messageHandler
            )
            return job.callback(message.result)
        }
    }

    worker.on(
        'message',
        messageHandler
    )
}

export function initWorkers(threads) {
    for (let i = 0;  i < threads; ++i) {
        workerFarm.push(new Worker('./libs/utils/multithread/genericWorker.js', { env: SHARE_ENV }))

        const worker = workerFarm[i]
        worker.postMessage({
            type: 'initWorker',
            name: `Worker ${parseInt(i) + 1}`
        })

        workerReady[i] = true
    }

    log(`Initialised ${threads} workers!`)

    setInterval(
        scheduler,
        process.env.SCHEDULE_TIME ?? 50
    )

    log(`Initialised the scheduler!`)
}

export function spread(script, data) {
    return new Promise(
        async (resolve, reject) => {
            const jobID = randomUUID()

            scheduledJobs.push({
                script,
                data,
                jobID,
                callback: result => {
                    resolve(result)
                }
            })
        }
    )
}