From 392160fdbbb45a93ebc90d57577fb62227de4c5e Mon Sep 17 00:00:00 2001 From: schneefux Date: Sat, 1 Apr 2017 12:02:40 +0200 Subject: rewrite in node --- worker.js | 163 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 163 insertions(+) create mode 100644 worker.js (limited to 'worker.js') diff --git a/worker.js b/worker.js new file mode 100644 index 0000000..6d05e24 --- /dev/null +++ b/worker.js @@ -0,0 +1,163 @@ +#!/usr/bin/node +/* jshint esnext:true */ +'use strict'; + +var amqp = require("amqplib"), + Seq = require("sequelize"); + +var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", + DATABASE_URI = process.env.DATABASE_URI || "sqlite:///db.sqlite", + BATCHSIZE = process.env.PROCESSOR_BATCH || 50 * (1 + 5), // matches + players + teams + IDLE_TIMEOUT = process.env.PROCESSOR_IDLETIMEOUT || 500; // ms + +(async () => { + let seq = new Seq(DATABASE_URI), + model = require("../orm/model")(seq, Seq), + rabbit = await amqp.connect(RABBITMQ_URI), + ch = await rabbit.createChannel(); + + let queue = [], + timer = undefined; + + await seq.sync(); + + await ch.assertQueue("compile", {durable: true}); + // as long as the queue is filled, msg are not ACKed + // server sends as long as there are less than `prefetch` unACKed + await ch.prefetch(BATCHSIZE); + + ch.consume("compile", async (msg) => { + queue.push(msg); + + // fill queue until batchsize or idle + if (timer === undefined) + timer = setTimeout(process, IDLE_TIMEOUT) + if (queue.length == BATCHSIZE) + await process(); + }, { noAck: false }); + + async function process() { + console.log("compiling batch", queue.length); + + // clean up to allow processor to accept while we wait for db + let msgs = queue.slice(); + queue = []; + clearTimeout(timer); + timer = undefined; + + // BEGIN + let transaction = await seq.transaction({ autocommit: false }); + + // UPSERT + try { + // processor sends to queue with a custom "type" so compiler can filter + // m.content: player.api_id + let players = msgs.filter((m) => m.properties.type == "player").map((m) => JSON.parse(m.content)), + participants = msgs.filter((m) => m.properties.type == "participant").map((m) => JSON.parse(m.content)); + + await Promise.all(players.map(async (player) => { + let player_api_id = player.api_id, + player_ext = {}; + + player_ext.player_api_id = player_api_id; + //player_ext.series = "" + + // TODO parallelize + + player_ext.played = await model.Participant.count({ + where: { + player_api_id: player_api_id + } + }); + player_ext.wins = await model.Participant.count({ + where: { + player_api_id: player_api_id, + winner: true + } + }); + + // TODO maybe this can be done in fewer/combined/subqueries + let count_matches_where = async (where) => { + return (await model.Participant.findOne({ + where: where, + attributes: [[seq.fn("COUNT", "$roster.match$"), "count"]], + include: [ { + model: model.Roster, + attributes: [], + include: [ { + model: model.Match, + attributes: [] + } ] + } ] + })).get("count"); + }; + player_ext.played_casual = await count_matches_where({ + player_api_id: player_api_id, + "$roster.match.game_mode$": "casual" + }); + player_ext.played_ranked = await count_matches_where({ + player_api_id: player_api_id, + "$roster.match.game_mode$": "ranked" + }); + player_ext.wins_casual = await count_matches_where({ + player_api_id: player_api_id, + winner: true, + "$roster.match.game_mode$": "casual" + }); + player_ext.wins_ranked = await count_matches_where({ + player_api_id: player_api_id, + winner: true, + "$roster.match.game_mode$": "ranked" + }); + + await model.PlayerExt.upsert(player_ext, { + include: [ model.Participant ], + transaction: transaction + }); + })); + await Promise.all(participants.map(async (api_participant) => { + let participant = await model.Participant.findOne({ + where: { + api_id: api_participant.api_id + }, + attributes: ["api_id", "kills", "assists", "deaths", seq.col("roster.hero_kills")], + include: [ + model.Roster + ] + }), + participant_ext = {}; + + participant_ext.participant_api_id = participant.api_id; + participant_ext.series = "" // TODO rm + + if (participant.roster.hero_kills == 0) + participant_ext.kills_participation = 0; + else + participant_ext.kills_participation = (participant.kills + participant.assists) / participant.roster.hero_kills; + + if (participant.deaths == 0) + participant_ext.kda = 0; + else + participant_ext.kda = (participant.kills + participant.assists) / participant.deaths; + + await model.ParticipantExt.upsert(participant_ext, { + include: [ model.Participant ], + transaction: transaction + }); + })); + + // COMMIT + await transaction.commit(); + console.log("acking batch"); + await ch.ack(msgs.pop(), true); // ack all messages until the last + + // notify web + await Promise.all(players.map(async (p) => await ch.publish("amq.topic", p.name, new Buffer("compile_commit")) )); + } catch (err) { // TODO catch only SQL error, also catch errors in the promises + console.error(err); + await transaction.rollback(); + await ch.nack(msgs.pop(), true, true); // nack all messages until the last and requeue + // TODO don't requeue broken records + } + } +})(); -- cgit v1.3.1 From bbfc647cff598cda5d82d1baee2b3d97b4b551e1 Mon Sep 17 00:00:00 2001 From: schneefux Date: Sat, 1 Apr 2017 12:42:23 +0200 Subject: calculate player.last_match_created_date --- worker.js | 21 +++++++++++++++++++++ 1 file changed, 21 insertions(+) (limited to 'worker.js') diff --git a/worker.js b/worker.js index 6d05e24..79b385b 100644 --- a/worker.js +++ b/worker.js @@ -59,6 +59,27 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", let player_api_id = player.api_id, player_ext = {}; + // set last_match_created_date + let lmcd = (await model.Participant.findOne({ + where: { + player_api_id: player_api_id + }, + attributes: [ [seq.col("roster.match.created_at"), "last_match_created_date"] ], + include: [ { + model: model.Roster, + attributes: [], + include: [ { + model: model.Match, + attributes: [] + } ] + } ], + order: [ + [seq.col("last_match_created_date"), "DESC"] + ] + })).get("last_match_created_date"); + await model.Player.update({ last_match_created_date: lmcd }, { where: { api_id: player_api_id } }); + + // calculate "extended" player_ext fields like wins per patch player_ext.player_api_id = player_api_id; //player_ext.series = "" -- cgit v1.3.1 From 81ed09e1d200aaf85e9c7b4ecf3544597a3446ba Mon Sep 17 00:00:00 2001 From: schneefux Date: Sun, 2 Apr 2017 20:26:03 +0200 Subject: insert in batches --- worker.js | 107 ++++++++++++++++++++++++++++++++++++-------------------------- 1 file changed, 62 insertions(+), 45 deletions(-) (limited to 'worker.js') diff --git a/worker.js b/worker.js index 79b385b..066310d 100644 --- a/worker.js +++ b/worker.js @@ -45,17 +45,20 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", clearTimeout(timer); timer = undefined; - // BEGIN - let transaction = await seq.transaction({ autocommit: false }); - - // UPSERT - try { - // processor sends to queue with a custom "type" so compiler can filter - // m.content: player.api_id - let players = msgs.filter((m) => m.properties.type == "player").map((m) => JSON.parse(m.content)), - participants = msgs.filter((m) => m.properties.type == "participant").map((m) => JSON.parse(m.content)); - - await Promise.all(players.map(async (player) => { + // aggregate & bulk insert + let player_ext_records = [], + participant_ext_records = [], + player_updates = []; // [[what, where]] + + // processor sends to queue with a custom "type" so compiler can filter + // m.content: player.api_id + let players = msgs.filter((m) => m.properties.type == "player").map((m) => JSON.parse(m.content)), + participants = msgs.filter((m) => m.properties.type == "participant").map((m) => JSON.parse(m.content)); + + // collect information and populate _record arrays + await Promise.all([ + // collect player information + Promise.all(players.map(async (player) => { let player_api_id = player.api_id, player_ext = {}; @@ -77,7 +80,8 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", [seq.col("last_match_created_date"), "DESC"] ] })).get("last_match_created_date"); - await model.Player.update({ last_match_created_date: lmcd }, { where: { api_id: player_api_id } }); + // do later in the transaction + player_updates.push([{ last_match_created_date: lmcd }, { where: { api_id: player_api_id } }]); // calculate "extended" player_ext fields like wins per patch player_ext.player_api_id = player_api_id; @@ -85,18 +89,6 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", // TODO parallelize - player_ext.played = await model.Participant.count({ - where: { - player_api_id: player_api_id - } - }); - player_ext.wins = await model.Participant.count({ - where: { - player_api_id: player_api_id, - winner: true - } - }); - // TODO maybe this can be done in fewer/combined/subqueries let count_matches_where = async (where) => { return (await model.Participant.findOne({ @@ -112,6 +104,20 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", } ] })).get("count"); }; + + // TODO run in parallel + player_ext.played = await model.Participant.count({ + where: { + player_api_id: player_api_id + } + }); + player_ext.wins = await model.Participant.count({ + where: { + player_api_id: player_api_id, + winner: true + } + }); + player_ext.played_casual = await count_matches_where({ player_api_id: player_api_id, "$roster.match.game_mode$": "casual" @@ -131,12 +137,10 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", "$roster.match.game_mode$": "ranked" }); - await model.PlayerExt.upsert(player_ext, { - include: [ model.Participant ], - transaction: transaction - }); - })); - await Promise.all(participants.map(async (api_participant) => { + player_ext_records.push(player_ext); + })), + // calculate participant fields + Promise.all(participants.map(async (api_participant) => { let participant = await model.Participant.findOne({ where: { api_id: api_participant.api_id @@ -161,24 +165,37 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", else participant_ext.kda = (participant.kills + participant.assists) / participant.deaths; - await model.ParticipantExt.upsert(participant_ext, { - include: [ model.Participant ], - transaction: transaction - }); - })); - - // COMMIT - await transaction.commit(); - console.log("acking batch"); - await ch.ack(msgs.pop(), true); // ack all messages until the last + participant_ext_records.push(participant_ext); + })) + ]); - // notify web - await Promise.all(players.map(async (p) => await ch.publish("amq.topic", p.name, new Buffer("compile_commit")) )); - } catch (err) { // TODO catch only SQL error, also catch errors in the promises + // load records into db + try { + console.log("inserting batch into db"); + await seq.transaction({ autocommit: false }, (transaction) => { + return Promise.all([ + model.ParticipantExt.bulkCreate(participant_ext_records, { + updateOnDuplicate: [], // all + transaction: transaction + }), + model.PlayerExt.bulkCreate(player_ext_records, { + updateOnDuplicate: [], // all + transaction: transaction + }), + player_updates.map(async (pu) => + await model.Player.update(pu[0], pu[1])) + ]); + }); + } catch (err) { + // this should only happen for deadlocks or non-data related issues console.error(err); - await transaction.rollback(); await ch.nack(msgs.pop(), true, true); // nack all messages until the last and requeue - // TODO don't requeue broken records + return; // give up } + console.log("acking batch"); + await ch.ack(msgs.pop(), true); // ack all messages until the last + + // notify web + await Promise.all(players.map(async (p) => await ch.publish("amq.topic", p.name, new Buffer("compile_commit")) )); } })(); -- cgit v1.3.1 From fda35295a35e807709e1ef8b8d4dfb1e443f4d8a Mon Sep 17 00:00:00 2001 From: schneefux Date: Mon, 3 Apr 2017 16:05:25 +0200 Subject: redesign web notification --- worker.js | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) (limited to 'worker.js') diff --git a/worker.js b/worker.js index 066310d..652faf7 100644 --- a/worker.js +++ b/worker.js @@ -196,6 +196,7 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", await ch.ack(msgs.pop(), true); // ack all messages until the last // notify web - await Promise.all(players.map(async (p) => await ch.publish("amq.topic", p.name, new Buffer("compile_commit")) )); + await Promise.all(players.map(async (p) => await ch.publish("amq.topic", "player." + p.name, + new Buffer("stats_update")) )); } })(); -- cgit v1.3.1 From 3ab7ff81923d4472c6aa04e5a9a5f36e64d6fdd8 Mon Sep 17 00:00:00 2001 From: schneefux Date: Mon, 3 Apr 2017 16:33:28 +0200 Subject: do not log queries --- worker.js | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'worker.js') diff --git a/worker.js b/worker.js index 652faf7..0ab63e1 100644 --- a/worker.js +++ b/worker.js @@ -11,7 +11,7 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", IDLE_TIMEOUT = process.env.PROCESSOR_IDLETIMEOUT || 500; // ms (async () => { - let seq = new Seq(DATABASE_URI), + let seq = new Seq(DATABASE_URI, { logging: () => {} }), model = require("../orm/model")(seq, Seq), rabbit = await amqp.connect(RABBITMQ_URI), ch = await rabbit.createChannel(); -- cgit v1.3.1 From a41d397a57be9e7595baa9310f83d2d6a411175f Mon Sep 17 00:00:00 2001 From: schneefux Date: Mon, 3 Apr 2017 16:43:29 +0200 Subject: anticipate players with no match history --- worker.js | 18 ++++++++++++------ 1 file changed, 12 insertions(+), 6 deletions(-) (limited to 'worker.js') diff --git a/worker.js b/worker.js index 0ab63e1..585333a 100644 --- a/worker.js +++ b/worker.js @@ -63,7 +63,7 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", player_ext = {}; // set last_match_created_date - let lmcd = (await model.Participant.findOne({ + let record = await model.Participant.findOne({ where: { player_api_id: player_api_id }, @@ -79,9 +79,13 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", order: [ [seq.col("last_match_created_date"), "DESC"] ] - })).get("last_match_created_date"); - // do later in the transaction - player_updates.push([{ last_match_created_date: lmcd }, { where: { api_id: player_api_id } }]); + }); + if (record != null) + // do later in the transaction + player_updates.push([ + { last_match_created_date: record.get("last_match_created_date") }, + { where: { api_id: player_api_id } } + ]); // calculate "extended" player_ext fields like wins per patch player_ext.player_api_id = player_api_id; @@ -91,7 +95,7 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", // TODO maybe this can be done in fewer/combined/subqueries let count_matches_where = async (where) => { - return (await model.Participant.findOne({ + let record = await model.Participant.findOne({ where: where, attributes: [[seq.fn("COUNT", "$roster.match$"), "count"]], include: [ { @@ -102,7 +106,9 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", attributes: [] } ] } ] - })).get("count"); + }) + if (record == null) return 0; + return record.get("count"); }; // TODO run in parallel -- cgit v1.3.1 From b7b7528a0639e0ad4cd1bdac0d021752ddbf0401 Mon Sep 17 00:00:00 2001 From: schneefux Date: Tue, 4 Apr 2017 11:06:17 +0200 Subject: send notifications for participant stats --- worker.js | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) (limited to 'worker.js') diff --git a/worker.js b/worker.js index 585333a..11da36f 100644 --- a/worker.js +++ b/worker.js @@ -202,7 +202,11 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", await ch.ack(msgs.pop(), true); // ack all messages until the last // notify web - await Promise.all(players.map(async (p) => await ch.publish("amq.topic", "player." + p.name, - new Buffer("stats_update")) )); + await Promise.all([ + Promise.all(players.map(async (p) => await ch.publish("amq.topic", "player." + p.name, + new Buffer("stats_update")) )), + Promise.all(participant_ext_records.map(async (p) => await ch.publish("amq.topic", + "participant." + p.api_id, new Buffer("stats_update")) )) + ]); } })(); -- cgit v1.3.1 From 3406a3e76754859cbee257a61f150160597b73fa Mon Sep 17 00:00:00 2001 From: schneefux Date: Tue, 4 Apr 2017 12:06:26 +0200 Subject: wait for rabbit and db to start --- package.json | 3 ++- worker.js | 23 +++++++++++++++++------ 2 files changed, 19 insertions(+), 7 deletions(-) (limited to 'worker.js') diff --git a/package.json b/package.json index 0f84dca..66afc71 100644 --- a/package.json +++ b/package.json @@ -6,7 +6,8 @@ "dependencies": { "amqplib": "^0.5.1", "mysql": "^2.13.0", - "sequelize": "^3.30.4" + "sequelize": "^3.30.4", + "sleep-promise": "^2.0.0" }, "devDependencies": {}, "scripts": { diff --git a/worker.js b/worker.js index 11da36f..a31f97d 100644 --- a/worker.js +++ b/worker.js @@ -3,7 +3,8 @@ 'use strict'; var amqp = require("amqplib"), - Seq = require("sequelize"); + Seq = require("sequelize"), + sleep = require("sleep-promise"); var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", DATABASE_URI = process.env.DATABASE_URI || "sqlite:///db.sqlite", @@ -11,17 +12,27 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", IDLE_TIMEOUT = process.env.PROCESSOR_IDLETIMEOUT || 500; // ms (async () => { - let seq = new Seq(DATABASE_URI, { logging: () => {} }), - model = require("../orm/model")(seq, Seq), - rabbit = await amqp.connect(RABBITMQ_URI), - ch = await rabbit.createChannel(); + let seq, model, rabbit, ch; + + while (true) { + try { + seq = new Seq(DATABASE_URI, { logging: () => {} }), + rabbit = await amqp.connect(RABBITMQ_URI), + ch = await rabbit.createChannel(); + await ch.assertQueue("compile", {durable: true}); + break; + } catch (err) { + console.error(err); + await sleep(5000); + } + } + model = require("../orm/model")(seq, Seq); let queue = [], timer = undefined; await seq.sync(); - await ch.assertQueue("compile", {durable: true}); // as long as the queue is filled, msg are not ACKed // server sends as long as there are less than `prefetch` unACKed await ch.prefetch(BATCHSIZE); -- cgit v1.3.1 From d7fcdb2ee11fa2f5a10f911b6916741ea0672734 Mon Sep 17 00:00:00 2001 From: schneefux Date: Tue, 4 Apr 2017 17:21:31 +0200 Subject: fix player_ext notifications --- worker.js | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) (limited to 'worker.js') diff --git a/worker.js b/worker.js index a31f97d..6e4e4cb 100644 --- a/worker.js +++ b/worker.js @@ -217,7 +217,7 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", Promise.all(players.map(async (p) => await ch.publish("amq.topic", "player." + p.name, new Buffer("stats_update")) )), Promise.all(participant_ext_records.map(async (p) => await ch.publish("amq.topic", - "participant." + p.api_id, new Buffer("stats_update")) )) + "participant." + p.participant_api_id, new Buffer("stats_update")) )) ]); } })(); -- cgit v1.3.1 From de9bd96b5ccda3b9a354cf3e02d49012bdc177d4 Mon Sep 17 00:00:00 2001 From: schneefux Date: Tue, 4 Apr 2017 20:30:33 +0200 Subject: send jobs to compiler --- worker.js | 9 +++++++++ 1 file changed, 9 insertions(+) (limited to 'worker.js') diff --git a/worker.js b/worker.js index 6e4e4cb..d0d6ca0 100644 --- a/worker.js +++ b/worker.js @@ -20,6 +20,7 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", rabbit = await amqp.connect(RABBITMQ_URI), ch = await rabbit.createChannel(); await ch.assertQueue("compile", {durable: true}); + await ch.assertQueue("analyze", {durable: true}); break; } catch (err) { console.error(err); @@ -212,6 +213,14 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", console.log("acking batch"); await ch.ack(msgs.pop(), true); // ack all messages until the last + // notify analyzer + Promise.all(participant_ext_records.map(async (p) => + await ch.sendToQueue("analyze", new Buffer(p.participant_api_id), { + persistent: true, + type: "participant" + }) + )); + // notify web await Promise.all([ Promise.all(players.map(async (p) => await ch.publish("amq.topic", "player." + p.name, -- cgit v1.3.1 From a6b1c4139a485595fd858fa447d893725d43a015 Mon Sep 17 00:00:00 2001 From: schneefux Date: Tue, 4 Apr 2017 21:03:28 +0200 Subject: filter undefined or null objects --- worker.js | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) (limited to 'worker.js') diff --git a/worker.js b/worker.js index d0d6ca0..b277eca 100644 --- a/worker.js +++ b/worker.js @@ -64,8 +64,10 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", // processor sends to queue with a custom "type" so compiler can filter // m.content: player.api_id - let players = msgs.filter((m) => m.properties.type == "player").map((m) => JSON.parse(m.content)), - participants = msgs.filter((m) => m.properties.type == "participant").map((m) => JSON.parse(m.content)); + let players = msgs.filter((m) => m.properties.type == "player").map( + (m) => JSON.parse(m.content)).filter((p) => p != undefined), + participants = msgs.filter((m) => m.properties.type == "participant").map( + (m) => JSON.parse(m.content)).filter((p) => p != undefined); // collect information and populate _record arrays await Promise.all([ -- cgit v1.3.1 From 7d085bac26ca372a3c9bd097f7c7214af5fbb408 Mon Sep 17 00:00:00 2001 From: schneefux Date: Tue, 4 Apr 2017 21:04:41 +0200 Subject: ack each message individually --- worker.js | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) (limited to 'worker.js') diff --git a/worker.js b/worker.js index b277eca..4fcb0bf 100644 --- a/worker.js +++ b/worker.js @@ -209,11 +209,11 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", } catch (err) { // this should only happen for deadlocks or non-data related issues console.error(err); - await ch.nack(msgs.pop(), true, true); // nack all messages until the last and requeue + await Promise.all(msgs.map((m) => ch.nack(m, true)) ); // requeue return; // give up } console.log("acking batch"); - await ch.ack(msgs.pop(), true); // ack all messages until the last + await Promise.all(msgs.map((m) => ch.ack(m)) ); // notify analyzer Promise.all(participant_ext_records.map(async (p) => -- cgit v1.3.1 From 9567fcbd4d41f603fab53864008add884bd315de Mon Sep 17 00:00:00 2001 From: schneefux Date: Tue, 4 Apr 2017 21:05:09 +0200 Subject: await transaction --- worker.js | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) (limited to 'worker.js') diff --git a/worker.js b/worker.js index 4fcb0bf..1efcba7 100644 --- a/worker.js +++ b/worker.js @@ -192,8 +192,8 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", // load records into db try { console.log("inserting batch into db"); - await seq.transaction({ autocommit: false }, (transaction) => { - return Promise.all([ + await seq.transaction({ autocommit: false }, async (transaction) => { + await Promise.all([ model.ParticipantExt.bulkCreate(participant_ext_records, { updateOnDuplicate: [], // all transaction: transaction -- cgit v1.3.1 From 241a277b78d9b3f94ddee550e2d6e7f08fff2a95 Mon Sep 17 00:00:00 2001 From: schneefux Date: Fri, 7 Apr 2017 18:33:14 +0200 Subject: update for new schema --- worker.js | 124 +++++++++++++++----------------------------------------------- 1 file changed, 30 insertions(+), 94 deletions(-) (limited to 'worker.js') diff --git a/worker.js b/worker.js index 1efcba7..5c1c5bf 100644 --- a/worker.js +++ b/worker.js @@ -8,7 +8,7 @@ var amqp = require("amqplib"), var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", DATABASE_URI = process.env.DATABASE_URI || "sqlite:///db.sqlite", - BATCHSIZE = process.env.PROCESSOR_BATCH || 50 * (1 + 5), // matches + players + teams + BATCHSIZE = process.env.PROCESSOR_BATCH || 50 * (1 + 5), // matches + players IDLE_TIMEOUT = process.env.PROCESSOR_IDLETIMEOUT || 500; // ms (async () => { @@ -58,8 +58,7 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", timer = undefined; // aggregate & bulk insert - let player_ext_records = [], - participant_ext_records = [], + let participant_stats_records = [], player_updates = []; // [[what, where]] // processor sends to queue with a custom "type" so compiler can filter @@ -73,8 +72,7 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", await Promise.all([ // collect player information Promise.all(players.map(async (player) => { - let player_api_id = player.api_id, - player_ext = {}; + let player_api_id = player.api_id; // set last_match_created_date let record = await model.Participant.findOne({ @@ -100,92 +98,20 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", { last_match_created_date: record.get("last_match_created_date") }, { where: { api_id: player_api_id } } ]); - - // calculate "extended" player_ext fields like wins per patch - player_ext.player_api_id = player_api_id; - //player_ext.series = "" - - // TODO parallelize - - // TODO maybe this can be done in fewer/combined/subqueries - let count_matches_where = async (where) => { - let record = await model.Participant.findOne({ - where: where, - attributes: [[seq.fn("COUNT", "$roster.match$"), "count"]], + })), + // calculate participant fields + Promise.all(participants.map(async (api_participant) => { + participant_stats_records.push(calculate_stats( + await model.Participant.findOne({ + where: { api_id: api_participant.api_id }, include: [ { model: model.Roster, - attributes: [], - include: [ { - model: model.Match, - attributes: [] - } ] + include: [ + model.Match + ] } ] }) - if (record == null) return 0; - return record.get("count"); - }; - - // TODO run in parallel - player_ext.played = await model.Participant.count({ - where: { - player_api_id: player_api_id - } - }); - player_ext.wins = await model.Participant.count({ - where: { - player_api_id: player_api_id, - winner: true - } - }); - - player_ext.played_casual = await count_matches_where({ - player_api_id: player_api_id, - "$roster.match.game_mode$": "casual" - }); - player_ext.played_ranked = await count_matches_where({ - player_api_id: player_api_id, - "$roster.match.game_mode$": "ranked" - }); - player_ext.wins_casual = await count_matches_where({ - player_api_id: player_api_id, - winner: true, - "$roster.match.game_mode$": "casual" - }); - player_ext.wins_ranked = await count_matches_where({ - player_api_id: player_api_id, - winner: true, - "$roster.match.game_mode$": "ranked" - }); - - player_ext_records.push(player_ext); - })), - // calculate participant fields - Promise.all(participants.map(async (api_participant) => { - let participant = await model.Participant.findOne({ - where: { - api_id: api_participant.api_id - }, - attributes: ["api_id", "kills", "assists", "deaths", seq.col("roster.hero_kills")], - include: [ - model.Roster - ] - }), - participant_ext = {}; - - participant_ext.participant_api_id = participant.api_id; - participant_ext.series = "" // TODO rm - - if (participant.roster.hero_kills == 0) - participant_ext.kills_participation = 0; - else - participant_ext.kills_participation = (participant.kills + participant.assists) / participant.roster.hero_kills; - - if (participant.deaths == 0) - participant_ext.kda = 0; - else - participant_ext.kda = (participant.kills + participant.assists) / participant.deaths; - - participant_ext_records.push(participant_ext); + )) })) ]); @@ -194,11 +120,7 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", console.log("inserting batch into db"); await seq.transaction({ autocommit: false }, async (transaction) => { await Promise.all([ - model.ParticipantExt.bulkCreate(participant_ext_records, { - updateOnDuplicate: [], // all - transaction: transaction - }), - model.PlayerExt.bulkCreate(player_ext_records, { + model.ParticipantStats.bulkCreate(participant_stats_records, { updateOnDuplicate: [], // all transaction: transaction }), @@ -216,7 +138,7 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", await Promise.all(msgs.map((m) => ch.ack(m)) ); // notify analyzer - Promise.all(participant_ext_records.map(async (p) => + Promise.all(participant_stats_records.map(async (p) => await ch.sendToQueue("analyze", new Buffer(p.participant_api_id), { persistent: true, type: "participant" @@ -227,8 +149,22 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", await Promise.all([ Promise.all(players.map(async (p) => await ch.publish("amq.topic", "player." + p.name, new Buffer("stats_update")) )), - Promise.all(participant_ext_records.map(async (p) => await ch.publish("amq.topic", + Promise.all(participant_stats_records.map(async (p) => await ch.publish("amq.topic", "participant." + p.participant_api_id, new Buffer("stats_update")) )) ]); } + + // based on the participant db record from the end of the match, + // calculate a participant_stats record and return it + function calculate_stats(participant) { + if (participant == null) { console.error("got nonexisting participant!"); return; } + let participant_stats = {}; + participant_stats.participant_api_id = participant.get("api_id"); + + participant_stats.kills = participant.get("kills"); + + participant_stats.kill_participation = participant.get("kills") / participant.roster.get("hero_kills"); + + return participant_stats; + } })(); -- cgit v1.3.1 From 72165f52411deaf633a7d6c301f0650c94f79671 Mon Sep 17 00:00:00 2001 From: schneefux Date: Fri, 7 Apr 2017 20:23:39 +0200 Subject: fix NaN stat --- worker.js | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) (limited to 'worker.js') diff --git a/worker.js b/worker.js index 5c1c5bf..84203ec 100644 --- a/worker.js +++ b/worker.js @@ -163,7 +163,8 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", participant_stats.kills = participant.get("kills"); - participant_stats.kill_participation = participant.get("kills") / participant.roster.get("hero_kills"); + if (participant.roster.get("hero_kills") == 0) participant_stats.kill_participation = 0; + else participant_stats.kill_participation = participant.get("kills") / participant.roster.get("hero_kills"); return participant_stats; } -- cgit v1.3.1 From 9932435c28a82164a02f3039b46dd5fb98d5223a Mon Sep 17 00:00:00 2001 From: schneefux Date: Fri, 7 Apr 2017 21:34:31 +0200 Subject: link items to participant --- worker.js | 23 ++++++++++++++++++----- 1 file changed, 18 insertions(+), 5 deletions(-) (limited to 'worker.js') diff --git a/worker.js b/worker.js index 84203ec..0c407eb 100644 --- a/worker.js +++ b/worker.js @@ -105,11 +105,16 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", await model.Participant.findOne({ where: { api_id: api_participant.api_id }, include: [ { - model: model.Roster, - include: [ - model.Match - ] - } ] + model: model.Roster, + include: [ + model.Match + ] + }, { + model: model.ItemParticipant, + as: "items", + include: [ model.Item ] + } + ] }) )) })) @@ -166,6 +171,14 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI || "amqp://localhost", if (participant.roster.get("hero_kills") == 0) participant_stats.kill_participation = 0; else participant_stats.kill_participation = participant.get("kills") / participant.roster.get("hero_kills"); + participant_stats.sustain_score = participant.items.reduce((score, item) => { + if (item.get("action") == "final") { + if (["Eve of Harvest", "Serpent Mask"].indexOf(item.item.get("name")) != -1) + return score + 20; + } + return score; + }, 0); + return participant_stats; } })(); -- cgit v1.3.1