From dd5a212422effb0a40bf346e7fe590e5521f4fdc Mon Sep 17 00:00:00 2001 From: undefined Date: Fri, 26 Jun 2020 18:30:43 +0800 Subject: [PATCH] =?UTF-8?q?=E5=A4=9Aworker=E5=85=B1=E4=BA=AB=E4=BA=8B?= =?UTF-8?q?=E4=BB=B6=EF=BC=88=E4=B8=8D=E7=A8=B3=E5=AE=9A?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- hydro/handler/domain.js | 2 +- hydro/handler/home.js | 4 +-- hydro/handler/manage.js | 2 +- hydro/handler/record.js | 8 ++--- hydro/model/task.js | 12 +++++++- hydro/model/user.js | 7 +++-- hydro/service/bus.js | 53 +++++++++++++++++++++++++++++----- hydro/utils.js | 5 ++++ module/migrate-vijos/script.js | 40 ++++++++++++++++++++++++- 9 files changed, 113 insertions(+), 20 deletions(-) diff --git a/hydro/handler/domain.js b/hydro/handler/domain.js index 2c4fcc867..5c4ae6d9d 100644 --- a/hydro/handler/domain.js +++ b/hydro/handler/domain.js @@ -73,7 +73,7 @@ class DomainEditHandler extends ManageHandler { async post(args) { const $set = {}; - for (const key of args) { + for (const key in args) { if (DOMAIN_SETTINGS_BY_KEY[key]) $set[key] = args[key]; } await domain.edit(args.domainId, $set); diff --git a/hydro/handler/home.js b/hydro/handler/home.js index b376b4728..50a99f223 100644 --- a/hydro/handler/home.js +++ b/hydro/handler/home.js @@ -272,7 +272,7 @@ class HomeMessagesHandler extends Handler { class HomeMessagesConnectionHandler extends ConnectionHandler { async prepare() { - bus.subscribe([`user_message-${this.user._id}`], (...args) => this.onMessageReceived(...args)); + bus.subscribe([`user_message-${this.user._id}`], this, 'onMessageReceived'); } async onMessageReceived(e) { @@ -280,7 +280,7 @@ class HomeMessagesConnectionHandler extends ConnectionHandler { } async cleanup() { - bus.unsubscribe(this.onMessageReceived); + bus.unsubscribe([`user_message-${this.user._id}`], this, 'onMessageReceived'); } } diff --git a/hydro/handler/manage.js b/hydro/handler/manage.js index b7996491b..803adfda9 100644 --- a/hydro/handler/manage.js +++ b/hydro/handler/manage.js @@ -106,7 +106,7 @@ class SystemScriptHandler extends SystemHandler { }); }, 500); this.response.body = { rid }; - this.response.redirect = this.url('record_main', { rid }); + this.response.redirect = this.url('record_detail', { rid }); } } diff --git a/hydro/handler/record.js b/hydro/handler/record.js index f5b9db683..c8f3e3a17 100644 --- a/hydro/handler/record.js +++ b/hydro/handler/record.js @@ -103,7 +103,7 @@ class RecordMainConnectionHandler extends RecordConnectionHandler { return; } } - bus.subscribe(['record_change'], this.onRecordChange); + bus.subscribe(['record_change'], this, 'onRecordChange'); } async message(msg) { @@ -118,7 +118,7 @@ class RecordMainConnectionHandler extends RecordConnectionHandler { } async cleanup() { - bus.unsubscribe(['record_change'], this.onRecordChange); + bus.unsubscribe(['record_change'], this, 'onRecordChange'); } async onRecordChange(data) { @@ -145,7 +145,7 @@ class RecordDetailConnectionHandler extends contest.ContestHandlerMixin(Connecti } } this.rid = rid; - bus.subscribe(['record_change'], this.onRecordChange); + bus.subscribe(['record_change'], this, 'onRecordChange'); this.onRecordChange({ value: rdoc }); } @@ -159,7 +159,7 @@ class RecordDetailConnectionHandler extends contest.ContestHandlerMixin(Connecti } async cleanup() { - bus.unsubscribe(['record_change'], this.onRecordChange); + bus.unsubscribe(['record_change'], this, 'onRecordChange'); } } diff --git a/hydro/model/task.js b/hydro/model/task.js index 4f5c95fa9..79f3bb3fa 100644 --- a/hydro/model/task.js +++ b/hydro/model/task.js @@ -1,3 +1,4 @@ +const cluster = require('cluster'); const moment = require('moment-timezone'); const db = require('../service/db'); @@ -6,6 +7,8 @@ const coll = db.collection('task'); async function add(task) { const t = { ...task }; if (typeof t.executeAfter === 'object') t.executeAfter = t.executeAfter.getTime(); + t.count = t.count || 1; + t.wait = t.wait || Object.keys(cluster.workers); t.executeAfter = t.executeAfter || new Date().getTime(); const res = await coll.insertOne(t); return res.insertedId; @@ -26,9 +29,16 @@ function del(_id) { async function getFirst(query) { const q = { ...query }; q.executeAfter = q.executeAfter || { $lt: new Date().getTime() }; + q.wait = { $elemMatch: { $eq: cluster.worker.id } }; const res = await coll.find(q).sort('_id', 1).limit(1).toArray(); if (res.length) { - await coll.deleteOne({ _id: res[0]._id }); + if (res[0].count === 1) await coll.deleteOne({ _id: res[0]._id }); + else { + await coll.updateOne( + { _id: res[0]._id }, + { $inc: { count: -1 }, $pull: { wait: cluster.worker.id } }, + ); + } if (res[0].interval) { await coll.insertOne({ ...res[0], executeAfter: moment().add(...res[0].interval).toDate(), diff --git a/hydro/model/user.js b/hydro/model/user.js index 96627d59e..c110990e8 100644 --- a/hydro/model/user.js +++ b/hydro/model/user.js @@ -109,8 +109,11 @@ function setPassword(uid, password) { ); } -function setById(uid, args) { - return coll.findOneAndUpdate({ _id: uid }, { $set: args }); +function setById(uid, $set, $unset) { + const op = {}; + if ($set && Object.keys($set).length) op.$set = $set; + if ($unset && Object.keys($unset).length) op.$unset = $unset; + return coll.findOneAndUpdate({ _id: uid }, op); } function setEmail(uid, mail) { diff --git a/hydro/service/bus.js b/hydro/service/bus.js index 2ba23c308..ec476fc49 100644 --- a/hydro/service/bus.js +++ b/hydro/service/bus.js @@ -1,15 +1,52 @@ const { EventEmitter } = require('events'); +const cluster = require('cluster'); const bus = new EventEmitter(); -global.Hydro.service.bus = module.exports = { - subscribe(events, handler) { +function subscribe(events, handler, funcName) { + if (!funcName) { for (const event of events) bus.on(event, handler); - }, - unsubscribe(events, handler) { + } else { + handler.__bus = (...args) => { + handler[funcName].call(handler, ...args); + }; + for (const event of events) bus.on(event, handler.__bus); + } +} + +function unsubscribe(events, handler, funcName) { + if (!funcName) { for (const event of events) bus.off(event, handler); - }, - publish(event, data) { - bus.emit(event, { value: data, event }); - }, + } else { + handler.__bus = (...args) => { + handler[funcName].call(handler, ...args); + }; + for (const event of events) bus.off(event, handler.__bus); + delete handler.__bus; + } +} + +function publish(event, data, isMaster = true) { + bus.emit(event, { value: data, event }); + if (isMaster && global.Hydro.model.task) { + global.Hydro.model.task.add({ + type: 'bus', + count: Object.keys(cluster.workers).length, + sender: cluster.worker.id, + event, + data, + }); + } +} + +function postInit() { + const { task } = global.Hydro.model; + task.consume({ type: 'bus' }, (data) => { + if (data.sender === cluster.worker.id) return; + publish(data.event, data.value, false); + }); +} + +global.Hydro.service.bus = module.exports = { + subscribe, unsubscribe, publish, postInit, }; diff --git a/hydro/utils.js b/hydro/utils.js index 3c1b70974..02cb9b2dc 100644 --- a/hydro/utils.js +++ b/hydro/utils.js @@ -1,10 +1,15 @@ const fs = require('fs'); +const cluster = require('cluster'); const path = require('path'); const superagent = require('superagent'); const proxy = require('superagent-proxy'); proxy(superagent); +if (!cluster.worker) { + cluster.worker = { id: 0 }; +} + const dict = 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ1234567890'; String.random = function random(digit = 32) { diff --git a/module/migrate-vijos/script.js b/module/migrate-vijos/script.js index 2de4e920d..c55669be5 100644 --- a/module/migrate-vijos/script.js +++ b/module/migrate-vijos/script.js @@ -2,7 +2,7 @@ const fs = require('fs'); const os = require('os'); -const { mongodb } = global.nodeModules; +const { mongodb, bson } = global.nodeModules; const dst = global.Hydro.service.db; const { file, problem, discussion } = global.Hydro.model; const { readConfig } = global.Hydro.lib; @@ -378,6 +378,42 @@ async function fixProblem(report) { } } +/** + * @param {Date} ts + */ +function objid(ts) { + const p = Math.floor(ts.getTime() / 1000).toString(16); + const id = new bson.ObjectID(); + return p + id.toString().slice(8, 8 + 6 + 4 + 6); +} + +/** + * @param {import('mongodb').Db} src + * @param {*} report + */ +async function message(src, report) { + const count = await src.collection('message').find().count(); + await report({ progress: 1, message: `Messages: ${count}` }); + const total = Math.floor(count / 50); + for (let i = 0; i <= total; i++) { + const docs = await dst.collection('message') + .find().skip(i * 50).limit(50) + .toArray(); + for (const doc of docs) { + for (const msg of doc.reply) { + await dst.collection('message').insertOne({ + _id: objid(msg.at), + from: msg.sender_uid, + to: msg.sender_uid === doc.sender_uid ? doc.sendee_uid : doc.sender_uid, + content: msg.content, + unread: msg.status !== 0, + }); + } + } + await report({ progress: Math.round(100 * ((i + 1) / (total + 1))) }); + } +} + async function task(name, src, report) { const count = await cursor[name](src).count(); await report({ progress: 1, message: `${name}: ${count}` }); @@ -482,11 +518,13 @@ async function migrateVijos({ await dst.collection('system').insertOne({ _id: 'migrateVijosFs', value: true }); } await dst.collection('user').deleteMany({ _id: { $nin: [0, 1] } }); + await dst.collection('message').deleteMany(); const d = ['user', 'document', 'document.status', 'record', 'file']; for (const i of d) await task(i, src, report); await fixProblem(report); await discussionNode(src, report); await domainUser(src, report); + await message(src, report); return true; }