From aff563f77d7a6533ada36757e77879a0b996e32d Mon Sep 17 00:00:00 2001 From: schneefux Date: Sat, 8 Apr 2017 23:03:39 +0200 Subject: tweak batch params; change timeout behaviour; fix race condition --- worker.js | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) (limited to 'worker.js') diff --git a/worker.js b/worker.js index 8dcbf26..fd9a138 100644 --- a/worker.js +++ b/worker.js @@ -11,8 +11,8 @@ var amqp = require("amqplib"), var RABBITMQ_URI = process.env.RABBITMQ_URI, DATABASE_URI = process.env.DATABASE_URI, - BATCHSIZE = parseInt(process.env.PROCESSOR_BATCH) || 50 * (1 + 5), // matches + players - IDLE_TIMEOUT = parseFloat(process.env.PROCESSOR_IDLETIMEOUT) || 500; // ms + BATCHSIZE = parseInt(process.env.BATCHSIZE) || 2 * 50 * (1 + 5), // matches + players + IDLE_TIMEOUT = parseFloat(process.env.IDLE_TIMEOUT) || 1000; // ms (async () => { let seq, model, rabbit, ch; @@ -64,14 +64,14 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI, // server sends as long as there are less than `prefetch` unACKed await ch.prefetch(BATCHSIZE); - ch.consume("process", async (msg) => { + ch.consume("process", (msg) => { queue.push(msg); // fill queue until batchsize or idle - if (timer === undefined) - timer = setTimeout(process, IDLE_TIMEOUT); + if (timer != undefined) clearTimeout(timer); + timer = setTimeout(process, IDLE_TIMEOUT); if (queue.length == BATCHSIZE) - await process(); + process(); }, { noAck: false }); async function process() { @@ -233,9 +233,9 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI, // now access db try { - console.log("inserting batch into db"); // upsert whole batch in parallel await seq.transaction({ autocommit: false }, async (transaction) => { + console.log("inserting batch into db"); await Promise.all([ model.Match.bulkCreate(match_records, { include: [ model.Roster, model.Asset ], -- cgit v1.3.1