Batch processing steps:
- When a job is posted, push it to the
queue
. - Get the required number of items from the queue and put it in the
batch
. - If a batch was processed before this, wait till
BATCH_INTERVAL
time has passed. - Process the batch.
- Clear the batch.
- Monitor the queue. Back to step 2.
batcher.js
const QUEUE_MONITORING_INTERVAL = 200;
const BATCH_SIZE = 3;
const BATCH_INTERVAL = 1000;
const PROCESSING = "processing";
const IDLE = "idle";
const queue = [];
const batch = [];
let queueInterval = null;
let lastBatchProcessing = null;
let batchStatus = IDLE;
function pushJob(fn) {
const newPromise = new Promise((resolve, reject) => {
const job = createJob(fn, resolve, reject);
queue.push(job);
});
monitorQueue();
return newPromise;
}
function createJob(fn, resolve, reject) {
return {
fn,
resolve,
reject,
};
}
function monitorQueue() {
if (queueInterval == null) {
queueInterval = setInterval(controller, QUEUE_MONITORING_INTERVAL);
}
}
function controller() {
handleQueue();
handleBatch();
}
function handleQueue() {
if (batchStatus !== IDLE) {
return;
} else if (shouldStopMonitoringQueue()) {
stopMonitoringQueue();
} else if (queue.length && batch.length < BATCH_SIZE) {
pushToBatch();
}
}
function shouldStopMonitoringQueue() {
return (
queue.length === 0 &&
batch.length === 0 &&
Date.now() - lastBatchProcessing > BATCH_INTERVAL
);
}
function stopMonitoringQueue() {
clearInterval(queueInterval);
queueInterval = null;
}
function pushToBatch() {
const spaceRemaining = BATCH_SIZE - batch.length;
const jobsToPop = Math.min(queue.length, spaceRemaining);
const jobs = queue.splice(0, jobsToPop);
batch.push(...jobs);
}
function handleBatch() {
if (shouldProcessBatch()) {
processBatch();
}
}
function shouldProcessBatch() {
if (!(batch.length && batchStatus === IDLE)) {
return false;
}
return (
lastBatchProcessing == null ||
Date.now() - lastBatchProcessing > BATCH_INTERVAL
);
}
async function processBatch() {
batchStatus = PROCESSING;
const promises = getPromisesFromBatch();
const results = await Promise.allSettled(promises);
processResults(results);
clearBatch();
batchStatus = IDLE;
lastBatchProcessing = Date.now();
}
function getPromisesFromBatch() {
const promises = [];
for (const job of batch) {
const promise = Promise.resolve(job.fn());
promises.push(promise);
}
return promises;
}
function processResults(results) {
for (let i = 0; i < results.length; i++) {
const result = results[i];
const job = batch[i];
if (result.status === "fulfilled") {
job.resolve(result.value);
} else {
job.reject(result.reason);
}
}
}
function clearBatch() {
batch.splice(0, batch.length);
}
module.exports = {
pushJob,
};
test.js
const batcher = require("./batcher");
async function main() {
const start = Date.now();
const promises = [];
for (let i = 0; i < 20; i++) {
const promise = batcher.pushJob(async () => await fn());
promises.push(promise);
}
const results = await Promise.all(promises);
const timeElapsed = (Date.now() - start) / 1000;
console.log(results);
console.log("Time elapsed: ", timeElapsed);
}
async function fn() {
return new Promise((resolve) => setTimeout(() => resolve(Math.random()), 3000));
}
main();