Max Job Running

[jkevinturado]

HI,

So, I've been using these BullMQ for months now. But I've been wondering if I could increase the job that can run simultaneously. Right now I can run 8 jobs at the same time (1 queue and 1 worker). My machine got 4 cores and 8 threads and also runs in cluster mode using pm2.

Any idea?

Below is the sample code:

// Redis connection
const CONFIG_REDIS =
  env === 'development'
    ? {
        redisConn_DB: {
          connection: {
            host: process.env.REDIS_TEST_ENDPOINT,
            port: process.env.REDIS_TEST_PORT,
            password: process.env.REDIS_TEST_PASSWORD,
            enableOfflineQueue: false,
            connectionTimeout: 5000,
            lazyConnect: true,
            retryStrategy(times) {
              console.warn(`Retrying redis connection: attempt ${times}`);
              if (times === 10) {
                return new Error('Redis retry time Error');
              }
              return Math.min(times * 500, 2000);
            },
          },
        },
    }

// Queue
const fileprocessQueue = new QueueMQ(
  'fileprocessQueue',
  CONFIG_REDIS.redisConn_DB
);

// Add job to Queue
await fileprocessQueue.add('fileprocessQueue', {
  fileId,
  userEmail: user[0].email,
});

// Worker
const fileJobWorker = new Worker(
  'fileprocessQueue',
  fileProcessWorkerHandler,
  CONFIG_REDIS.redisConn_DB
);

// Worker Handler
const fileProcessWorkerHandler= async (job, token) => {
  console.log(`jobname: ${job.name}`);
  switch (job.name) {
    case 'fileprocessQueue': {

      const { fileId, userId } = job.data;
      const reqdata = {
        processid: job.id,
        fileId,
        userId,
        job,
      };

      await FileModel.update(
        {
          file_process_id: job.id,
        },
        { where: { id: fileId } }
      );

      await ProcessFile(reqdata);
      break;
    }
  
    default:
      break;
  }
};

[jkevinturado]

UP

[jkevinturado]

Here's the code that I run. I wrote a minimal base on your comment since this is a highly intensive code. The jobsEvents, database, and BullBoard are included, this is how I observe how the job flows.

//app.js

import { Queue as QueueMQ, Worker, QueueEvents } from 'bullmq';
import { createBullBoard } from '@bull-board/api';
import { BullMQAdapter } from '@bull-board/api/bullMQAdapter.js';
import { ExpressAdapter } from '@bull-board/express';

import { ProcessFile } from './controller/fileprocess.js';
import { CONFIG_REDIS } from './config.js';


const JobEvents = async () => {
  try {
    //active
    fileJobWorker.on('active', async (job) => {
      console.log(
        `active job id: ${job.id}, userId: ${job.data.userEmail}, fileId: ${job.data.fileId} `
      );
    });

    //progress
    fileJobWorker.on('progress', async (job, progress) => {
      console.log(
        `progress: ${progress},  id: ${job.id}, userId: ${job.data.userEmail}, fileId: ${job.data.fileId} `
      );
    });

    //complete
    fileJobWorker.on('completed', async (job) => {
      console.log(
        `completed job id: ${job.id}, userId: ${job.data.userEmail}, fileId: ${job.data.fileId} `
      );
    });

    fileJobWorker.on('failed', async (job) => {
      console.log(
        `failed job id: ${job.id}, userId: ${job.data.userEmail}, fileId: ${job.data.fileId} `
      );
    });
  } catch (error) {
    console.log(error);
    const err = new Error(error);
    err.statusCode(500);
    throw error;
  }
};

// worker handler
const fileProcessWorkerHandler = async (job, token) => {
  switch (job.name) {
    case 'testFileprocessQueue': {
      const { fileId, userId } = job.data;
      const reqdata = {
        processid: job.id,
        fileId,
        userId,
        job,
      };

      await ProcessFile(reqdata);
      break;
    }
    default:
      break;
  }
};

// Job events
const queueEvents = new QueueEvents(
  'fileprocessQueue',
  CONFIG_REDIS.redisConn_DB.queue
);

// queue
const fileprocessQueue = new QueueMQ(
  'fileprocessQueue',
  CONFIG_REDIS.redisConn_DB.queue
);

//worker
const fileJobWorker = new Worker(
  'fileprocessQueue',
  fileProcessWorkerHandler,
  CONFIG_REDIS.redisConn_DB.worker
);

const serverAdapter = new ExpressAdapter();
serverAdapter.setBasePath('/admin/queues');
const { addQueue, removeQueue, setQueues, replaceQueues } = createBullBoard({
  queues: [new BullMQAdapter(fileprocessQueue)],
  serverAdapter: serverAdapter,
});

//route
app.use(bodyparser.json());
app.use('/admin/queues', serverAdapter.getRouter());

app.post('/api/testaddbulkjob', async (req, res, next) => {
  try {
    // console.log(req);
    const { items } = req.body;

    items.forEach(async ({ fileId, userEmail }) => {
      await fileprocessQueue.add('testFileprocessQueue', {
        fileId,
        userEmail,
      });
    });
    res.status(200).json({ msg: 'test' });
  } catch (error) {
    next(error);
  }
});

try {
  JobEvents();
  app.listen(process.env.PORT || 4001);
  console.log(
    `Cluster Server running on PORT ${process.env.PORT} | Env: ${process.env.ENV}`
  );
} catch (err) {
  console.log(err);
}

//fileprocess.js

export const ProcessFile = async (reqdata) => {
  const tablecolumns = ['return', 'result', 'return_long', 'remove_reason_l'];
  const { fileId, job } = reqdata;

  try {
    const columns = tablecolumns.map(
      (column) => `\`${column.toLowerCase().trim()}\` varchar(150)`
    );
    await job.updateProgress(33);

    const DropIfExistandQuery = `DROP TABLE IF EXISTS \`${fileId}\`;`;
    const CreateTableQuery = `CREATE TABLE \`${fileId}\` (${columns.toString()});`;

   //Execute DatabaseQuery
    await ExecuteQuery()

   //Update Query
    await job.updateProgress(66);

   // Create file and store to S3
    await job.updateProgress(100);

    return true;
  } catch (error) {
    throw error;
  }
};

[manast]

As I mentioned, you need to isolate the issue that you are claiming you have found. That means that if the issue is in Bull, you need to remove all third-party code, including express, postgress, bullboard, etc, in other words, zero dependencies.

[jkevinturado]

Hi, I've updated the code above

[manast]

I still see a lot of unrelated stuff...

  const tablecolumns = ['return', 'result', 'return_long', 'remove_reason_l'];
  const { fileId, job } = reqdata;

  try {
    const columns = tablecolumns.map(
      (column) => `\`${column.toLowerCase().trim()}\` varchar(150)`
    );
    await job.updateProgress(33);

    const DropIfExistandQuery = `DROP TABLE IF EXISTS \`${fileId}\`;`;
    const CreateTableQuery = `CREATE TABLE \`${fileId}\` (${columns.toString()});`;

   //Execute DatabaseQuery
    await ExecuteQuery()

   //Update Query
    await job.updateProgress(66);

   // Create file and store to S3
    await job.updateProgress(100);

Btw, this is a discussion, you should create an issue instead so it can be tracked in the case a fix is needed.

[jkevinturado]

This is noted. I'll be creating an issue soon with a repo. Thank you