bullmq locks thread?

[pcace]

Hi there, i am starting to get along with bullmq, but run into a problem where a bullmq child seems to lock my main thread in nodejs.

i have an app.ts wich basically looks like this: (there is also a function to get the data from the bull stack, but that seems to work just fine..)

import { selfserviceWorker } from './worker'
import { Queue } from 'bullmq'

const cueueName = 'jobque'
// ...
const jobque = new Queue(cueueName, {
  connection: {
    host: process.env.REDIS_SERVER?.toString(),
    port: parseInt(process.env.REDIS_PORT ? process.env.REDIS_PORT.toString() : '6379'),
    password: process.env.REDIS_PASSWORD?.toString(),
  },
})
// ....
const app = express() //.use(apiKeyAuth([process.env.API_KEY!]));
const port = 3000
app.listen(port, () => {
    //...
})
const addJob = async (request: Request, response: Response, next: NextFunction) => {
 // ...
    const job = await jobque.add(jobID, {
      workingfolder: workingfolder,
      query: query,
      output: output,
      bboxSize: bboxSize,
      lat: lat,
      lon: lon,
      projection: projection,
    })
    const state = await job.getState()
    response.status(200).send({ job: job?.name, status: state })
  }
selfserviceWorker.on('completed', async job => {
})
selfserviceWorker.on('failed', (job, err) => {
})
app.get('/add', addJob)

then there is a worker wich looks basically like that:

import { Worker } from "bullmq";
// ..

export const selfserviceWorker = new Worker(
  "jobque",
  async (job) => {
    const query = job.data.query;
    //...
    let result: JSON | undefined = undefined;
    // .. magic with numbers
    result = await getLOD2Data(query) // this takes a long time to run...
  // ...
    return { result: result };
  },
  {
    autorun: true,
    lockDuration: 600000,
    connection: {
      host: process.env.REDIS_SERVER?.toString(),
      port: parseInt(
        process.env.REDIS_PORT ? process.env.REDIS_PORT.toString() : "6379"
      ),
      password: process.env.REDIS_PASSWORD?.toString(),
    },
  }
);

inside getLOD2Data() is an async function wich also takes some time. it looks like this:

const getLOD2Data = async ( ....  ) => {
   const json =  await writeBuildingFoundationToCityJson( ... )
   .then(() => console.log("done")).catch(e => console.log(e)) // this takes a lot of time
    return json
}

so here comes my trouble: inside this writeBuildingFoundationToCityJson() function i need to calculate millions of points. and to not block the whole thread i thought its a good idea to also use a bullmq worker to let this run and not block everything else. but it still does, like i would not have used a worker.

the writeBuildingFoundationToCityJson() function looks like this:

import { Queue } from 'bullmq'
import { intersectionWorker } from './worker/intersectionWorker'

const intersectionJobque = new Queue('intersections', {
    connection: {
        host: process.env.REDIS_SERVER?.toString(),
        port: parseInt(process.env.REDIS_PORT ? process.env.REDIS_PORT.toString() : '6379'),
        password: process.env.REDIS_PASSWORD?.toString(),
    },
})

const runIntersectionServiceWorker = async (
    coordinatesList: { x: number; y: number; z: number }[],
): Promise<void> => {
    const intersectionJob = await intersectionJobque.add(
        '0',
        {coordinatesList: coordinatesList },
        { removeOnComplete: true, removeOnFail: true }
    )
    return new Promise((resolve, reject) => {
        intersectionWorker.on('completed', async job => {
            if (job.id === intersectionJob.id && job.name === intersectionJob.name) {
                job.remove()
                resolve()
            }
        })
        intersectionWorker.on('failed', (job, err) => {
            if (job?.id === intersectionJob.id && job?.name === intersectionJob.name) {
                job?.remove()
                reject()
            }
        })
    })
}

export const writeBuildingFoundationToCityJson = async ( ... ) => {
    let coordinatesList: { x: number; y: number; z: number }[] = [ ... ]
    await runIntersectionServiceWorker(coordinatesList)
}

and the intersection Worker Looks like this:

import { Worker } from 'bullmq'
import KDBush from 'kdbush'

const writeHeight = async (coordinatesList: any, folder: any) => {

    console.log("DEBUG UNRESPONSIVE!!!!!---> ")
    const coordinates = coordinatesList as { x: number; y: number; z: number }[]
    // do stuff with numbers and KDTrees in some functions
    const filebuffer = await kdTreesandNumbersandstuff(...)
        console.log("<------ DEBUG AGAIN RESPONSIVE!!!!! ")
        return writefile
    }

export const intersectionWorker = new Worker('intersections', async job => {
    const coordinates = job.data.coordinatesList as {
        x: number
        y: number
        z: number
    }[]
    const folder = job.data.workingfolder as string
    await writeHeight(coordinates, folder)
}, {    lockDuration: 600000,
})

i have marked in the last worker where the whole thing gets blocked / unresponsive.
What is the problem here - is it that i have two workers within each other? or am i not seeing something else?

if anyone has an idea what i am doing wrong please let me know!!
I hope this is not a too long question, but i just cannot figure out what s happening here. :/

Thanks a lot!!

[manast]

A Worker in BullMQ is not a thread, however, you can use a sandboxed processor, in this case, the worker will run in an isolated process and then it will not block your main process: https://docs.bullmq.io/guide/workers/sandboxed-processors

[pcace]

Hi, thank you so much for the clarification! That explains a lot. The Problem now is, that i just cannot get this to work as shown in the example files - this is the error i get:

26s 9:18:11:502   Cannot find module './dist/helper/lod2/worker/intersectionWorker.js'
Require stack:
- /home/user/git/projectRoot/node_modules/bullmq/dist/cjs/classes/child-processor.js
- /home/user/git/projectRoot/node_modules/bullmq/dist/cjs/classes/master.js
undefined

can the problem lay here in my tsconfig?:

{
  "compilerOptions": {
    "target": "es2017",
    "lib": [
      "es2019"
    ],
    "module": "ESNext",
    "rootDir": "src",
    "moduleResolution": "node",
    "outDir": "./dist",
    "esModuleInterop": true,
    "allowSyntheticDefaultImports": true,
    "forceConsistentCasingInFileNames": true,
    "strict": true,
    "noImplicitAny": true,
    "skipLibCheck": true
  }
}

in my app.ts i initialize the worker like so:

const intersectionJobque = new Worker('intersectionQueue','./src/helper/lod2/worker/intersectionWorker.ts', connection)
or alternatively :
const intersectionJobque = new Worker('intersectionQueue','./dist/helper/lod2/worker/intersectionWorker.js', connection)
both generate the same Error as shown above.

what is weird is that when i change the location of the worker to something definitely wrong i get:

Error: File ../src/helper/lod2/worker/intersectionWorker.ts does not exist
    at new Worker (/home/johannes/git/FFYselfservingAPI/node_modules/bullmq/src/classes/worker.ts:232:17)
    at file:///home/johannes/git/FFYselfservingAPI/src/app.ts:47:28
    at ModuleJob.run (node:internal/modules/esm/module_job:194:25)

so that means that my problem is probably not that i have set a wrong filename but rather something related to bullmq / ts ?

if someone has any idea how i could fix that issue, it would be great to hear!

Thank you so much!

[manast]

Here you go: https://blog.taskforce.sh/using-typescript-with-bullmq/

[pcace]

hi, thanks for your reply, i have read the link you posted, and changed my tsconfig to match the one provided in the repository (https://github.com/taskforcesh/bullmq-typescript):

{
  "compilerOptions": {
    "outDir": "dist",
    "target": "ES6",
    "esModuleInterop": true,
    "moduleResolution": "node",
    "module": "commonjs"
  },
  "include": ["./src/**/*"]
}

and the package.json to:

 "scripts": {
    "build": "tsc",
    "start": "node './dist/app.js'",
    "dev": "tsc-watch --onSuccess \"node ./dist/app.js\" | pino-pretty",
    "test": "ts-mocha ./tests/**/*.test.ts"
  },

it took a while to make everything else work accordingly ;) but it builds and starts without any error now.
BUT: the initial problem is still here:

17s 11:27:29:354   reason:  Cannot find module './dist/helper/lod2/worker/intersectionWorker.js'
Require stack:
- /home/user/git/project/node_modules/bullmq/dist/cjs/classes/child-processor.js
- /home/user/git/project/node_modules/bullmq/dist/cjs/classes/master.js
undefined

i did follow the github example to set up the worker.

if you have any other idea what i could check or do to make this work i am very happy to hear from you :)!

EDIT:
the intersectionWorker.js file looks like that:

"use strict";
var __awaiter = (this && this.__awaiter) || function (thisArg, _arguments, P, generator) {
    function adopt(value) { return value instanceof P ? value : new P(function (resolve) { resolve(value); }); }
    return new (P || (P = Promise))(function (resolve, reject) {
        function fulfilled(value) { try { step(generator.next(value)); } catch (e) { reject(e); } }
        function rejected(value) { try { step(generator["throw"](value)); } catch (e) { reject(e); } }
        function step(result) { result.done ? resolve(result.value) : adopt(result.value).then(fulfilled, rejected); }
        step((generator = generator.apply(thisArg, _arguments || [])).next());
    });
};
Object.defineProperty(exports, "__esModule", { value: true });
const writeHeight_1 = require("./writeHeight");
function default_1(job) {
    return __awaiter(this, void 0, void 0, function* () {
        console.log("##################################INSIDE WORKER");
        const coordinates = job.data.coordinatesList;
        const folder = job.data.workingfolder;
        console.log(folder, coordinates[0]);
        yield (0, writeHeight_1.writeHeight)(coordinates, folder);
    });
}
exports.default = default_1;

[pcace]

Hi, i did manage to get a littlebit further - by i have no idea what...
i am now getting this Error:

32s 13:44:46:262   DEBUG : Job Nr: 29 has failed! with this Reason: 
32s 13:44:46:262   reason:  write EPIPE

any idea what that means?

[manast]

Check the error stack for the job that failed for gaining more information.

[pcace]

Hi, thanks for your patience! i have found the problem. it was a problematic use of some fs functions (writing reading files within the worker).
i did not quite figure out why that was not working, but without the filesystem functions it runs the worker now!