summaryrefslogtreecommitdiff
path: root/worker.js
diff options
context:
space:
mode:
Diffstat (limited to 'worker.js')
-rw-r--r--worker.js201
1 files changed, 139 insertions, 62 deletions
diff --git a/worker.js b/worker.js
index fbced5f..68ad4cd 100644
--- a/worker.js
+++ b/worker.js
@@ -9,7 +9,7 @@ var amqp = require("amqplib"),
var RABBITMQ_URI = process.env.RABBITMQ_URI,
DATABASE_URI = process.env.DATABASE_URI,
- BATCHSIZE = parseInt(process.env.BATCHSIZE) || 5, // players + globals
+ BATCHSIZE = parseInt(process.env.BATCHSIZE) || 200, // players + globals
IDLE_TIMEOUT = parseFloat(process.env.IDLE_TIMEOUT) || 1000; // ms
(async () => {
@@ -29,73 +29,52 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI,
}
model = require("../orm/model")(seq, Seq);
- await seq.sync();
-
- await ch.prefetch(1); // TODO batching
-
- ch.consume("crunch", async (msg) => {
- if (msg.properties.type == "global") {}
-
- try {
- if (msg.properties.type == "player")
- await calculate_player_point(msg.content.toString());
- console.log("acking");
- await ch.ack(msg);
- } catch (err) {
- console.error(err);
- await ch.nack(msg, false, false); // nack and do not requeue
- }
- }, { noAck: false });
+ function cartesian(arr) {
+ return Array.prototype.reduce.call(arr, function(a, b) {
+ let ret = [];
+ a.forEach(function(a) {
+ b.forEach(function(b) {
+ ret.push(a.concat([b]));
+ });
+ });
+ return ret;
+ }, [[]]);
+ }
- async function calculate_player_point(player_api_id) {
- let win_rate, pick_rate, gold_per_min;
+ let dimension_cache = [];
+ // create a 3D array
+ // [[ ["hero", "Vox"], ["hero", "Taka"], …], [ ["game_mode", "ranked"], … ], …]
+ // (will not use "Vox" but the index instead)
+ await Promise.all(
+ [model.Series, model.Filter, model.Hero,
+ model.GameMode].map(async (d, idx) =>
+ dimension_cache[idx] =
+ (await d.findAll()).map((o) => [d, o]))
+ );
- let player = await model.Player.findOne(
- { where: { api_id: player_api_id } });
- // TODO cache dimensions
- // Series and Filter are special
- let dimensions = [model.Series, model.Filter, model.Hero, model.GameMode],
- player_dimensions = ["series", "filter", "hero", "game_mode"],
- dimension_cache = [];
- // create a 3D array
- // [[ ["hero", "Vox"], ["hero", "Taka"], …], [ ["game_mode", "ranked"], … ], …]
- // (will not use "Vox" but the index instead)
- await Promise.all(
- dimensions.map(async (d, idx) =>
- dimension_cache[idx] =
- (await d.findAll()).map((o) => [d, o]))
- );
+ // create an array with every possible combination
+ // hero x game mode x …
+ // Vox x casual x …
+ // SAW x casual x …
+ // …
+ // Vox x ranked x …
+ // SAW x ranked x …
+ // …
+ // Vox x ANY x …
+ let points = cartesian(dimension_cache);
- function cartesian(arr) {
- return Array.prototype.reduce.call(arr, function(a, b) {
- let ret = [];
- a.forEach(function(a) {
- b.forEach(function(b) {
- ret.push(a.concat([b]));
- });
- });
- return ret;
- }, [[]]);
- }
- // create an array with every possible combination
- // hero x game mode x …
- // Vox x casual x …
- // SAW x casual x …
- // …
- // Vox x ranked x …
- // SAW x ranked x …
- // …
- // Vox x ANY x …
- let combos = cartesian(dimension_cache);
- await Promise.all(combos.map(async (combo) => {
+ // return every possible [query, insert] combination
+ function calculate_point(dimensions) {
+ return points.map((point) => {
+ // Series and Filter are special
let where_aggr = {},
where_links = {};
// create skeleton: where hero_id=$hero
// for aggregation, use series as range
// for links, use the id
- combo.map((tuple) => {
+ point.forEach((tuple) => {
// [table name, table element]
- if (player_dimensions.indexOf(tuple[0].tableName) > -1) {
+ if (dimensions.indexOf(tuple[0].tableName) > -1) {
if (tuple[1].get("name") != "all") {
// exclude series & filter, added below
if (tuple[0].tableName == "filter") {
@@ -116,6 +95,106 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI,
where_links[tuple[0].tableName + "_id"] = tuple[1].id
}
});
+ return [where_aggr, where_links];
+ });
+ }
+ let player_points = calculate_point(["series", "filter", "hero", "game_mode"]),
+ global_points = calculate_point(["series", "filter", "hero", "game_mode"]); // TODO
+
+ let queue = [],
+ timer = undefined;
+
+ await seq.sync();
+
+ await ch.prefetch(BATCHSIZE);
+
+ ch.consume("crunch", (msg) => {
+ queue.push(msg);
+
+ // fill queue until batchsize or idle
+ if (timer != undefined) clearTimeout(timer);
+ timer = setTimeout(process, IDLE_TIMEOUT);
+ if (queue.length == BATCHSIZE)
+ process();
+ }, { noAck: false });
+
+ async function process() {
+ console.log("processing batch", queue.length);
+
+ // clean up to allow processor to accept while we wait for db
+ clearTimeout(timer);
+ timer = undefined;
+ let msgs = queue;
+ queue = [];
+
+ let player_records = [],
+ global_records = [];
+
+ await Promise.all(msgs.map(async (msg) => {
+ if (msg.properties.type == "global") {
+ // TODO
+ let stats = await calculate_global_point();
+ if (stats != undefined) global_records.push(stats);
+ }
+ if (msg.properties.type == "player") {
+ let stats = await calculate_player_point(
+ msg.content.toString());
+ if (stats != undefined) player_records.push(stats);
+ }
+ }));
+
+ try {
+ await seq.transaction({ autocommit: false }, async (transaction) => {
+ console.log("inserting batch into db");
+ await Promise.all([
+ model.PlayerPoint.bulkCreate(player_records, {
+ updateOnDuplicate: [], // all
+ transaction: transaction
+ }),
+ model.GlobalPoint.bulkCreate(global_records, {
+ updateOnDuplicate: [], // all
+ transaction: transaction
+ })
+ ]);
+ });
+ console.log("acking batch");
+ await Promise.all(msgs.map((m) => ch.ack(m)) );
+ } catch (err) {
+ // TODO
+ console.error(err);
+ await Promise.all(msgs.map((m) => ch.nack(m, true)) ); // requeue
+ }
+ }
+
+ async function calculate_global_point() {
+ console.log("crunching global stats, this could take a while");
+
+ await Promise.all(global_points.map(async (tuple) => {
+ let where_aggr = tuple[0],
+ where_links = tuple[1];
+ // aggregate participant_stats with our condition
+ let stats = await aggregate_stats(where_aggr);
+ if (stats != undefined) {
+ console.log("inserting global stats");
+ Object.assign(stats, where_links);
+ return stats;
+ }
+ return undefined;
+ }));
+ }
+
+ async function calculate_player_point(player_api_id) {
+ let player = await model.Player.findOne(
+ { where: { api_id: player_api_id } });
+ if (player == null) console.error("player does not exist in db!?");
+ if (player.get("last_update") == null)
+ // not enough data
+ return;
+ console.log("crunching player", player.name);
+
+ await Promise.all(player_points.map(async (tuple) => {
+ let where_aggr = tuple[0],
+ where_links = tuple[1];
// make it player specific
where_aggr["$participant.player_api_id$"] = player_api_id;
where_links["player_id"] = player.id;
@@ -124,9 +203,7 @@ var RABBITMQ_URI = process.env.RABBITMQ_URI,
if (stats != undefined) {
console.log("inserting stats for player", player.name);
Object.assign(stats, where_links);
- await model.PlayerPoint.upsert(stats, {
- where: where_links
- });
+ await model.PlayerPoint.upsert(stats);
}
}));
}
generated by cgit v1.3.1 (git 2.54.0) at 2026-08-02 04:56:16 +0000