From 2b4b7de28d48cf8511d9d9589a10a2d18ea48fcf Mon Sep 17 00:00:00 2001 From: undefined Date: Mon, 17 May 2021 23:56:34 +0800 Subject: [PATCH] core: add task manager --- package.json | 10 +- packages/hydrooj/package.json | 4 +- packages/hydrooj/src/handler/import.syzoj.ts | 110 +++++++------- packages/hydrooj/src/model/task.ts | 143 +++++++++++-------- packages/hydrooj/src/ui.ts | 44 +----- packages/ui-default/package.json | 12 +- 6 files changed, 152 insertions(+), 171 deletions(-) diff --git a/package.json b/package.json index 0f0253e80..314fead72 100644 --- a/package.json +++ b/package.json @@ -29,7 +29,7 @@ "@shelf/jest-mongodb": "^1.2.3", "@types/cross-spawn": "^6.0.2", "@types/jest": "^26.0.23", - "@types/node": "^15.0.2", + "@types/node": "^15.3.0", "@types/semver": "^7.3.5", "@typescript-eslint/eslint-plugin": "^4.22.1", "@typescript-eslint/parser": "^4.22.1", @@ -38,17 +38,17 @@ "eslint": "^7.26.0", "eslint-config-airbnb-typescript": "^12.0.0", "eslint-import-resolver-typescript": "^2.4.0", - "eslint-plugin-import": "^2.22.1", - "fs-extra": "^9.1.0", + "eslint-plugin-import": "^2.23.2", + "fs-extra": "^10.0.0", "globby": "^11.0.3", "jest": "^26.6.3", "latest-version": "^5.1.0", - "nmls": "^2.0.3", + "nmls": "^3.0.0", "ora": "^5.4.0", "semver": "^7.3.5", "ts-jest": "^26.5.6", "typedoc": "^0.20.36", "typescript": "4.2.4", - "yargs": "^16.2.0" + "yargs": "^17.0.1" } } diff --git a/packages/hydrooj/package.json b/packages/hydrooj/package.json index 067686a17..1a8076671 100644 --- a/packages/hydrooj/package.json +++ b/packages/hydrooj/package.json @@ -1,6 +1,6 @@ { "name": "hydrooj", - "version": "2.26.8", + "version": "2.26.9", "bin": "bin/hydrooj.js", "main": "dist/loader.js", "typings": "dist/loader.d.ts", @@ -17,7 +17,6 @@ "cookies": "^0.8.0", "detect-browser": "^5.1.1", "fs-extra": "^10.0.0", - "inquirer": "^8.0.0", "js-yaml": "^4.0.0", "koa": "^2.13.1", "koa-body": "^4.2.0", @@ -41,7 +40,6 @@ "devDependencies": { "@types/adm-zip": "^0.4.34", "@types/fs-extra": "^9.0.10", - "@types/inquirer": "^7.3.1", "@types/js-yaml": "^4.0.0", "@types/koa": "^2.13.1", "@types/koa-compress": "^4.0.1", diff --git a/packages/hydrooj/src/handler/import.syzoj.ts b/packages/hydrooj/src/handler/import.syzoj.ts index e42495866..37bec129b 100644 --- a/packages/hydrooj/src/handler/import.syzoj.ts +++ b/packages/hydrooj/src/handler/import.syzoj.ts @@ -11,6 +11,7 @@ import { ValidationError, RemoteOnlineJudgeError } from '../error'; import { Logger } from '../logger'; import type { ContentNode } from '../interface'; import problem from '../model/problem'; +import TaskModel from '../model/task'; import { PERM } from '../model/builtin'; import { Route, Handler, Types, post, @@ -21,6 +22,52 @@ import download from '../lib/download'; const RE_SYZOJ = /(https?):\/\/([^/]+)\/(problem|p)\/([0-9]+)\/?/i; const logger = new Logger('import.syzoj'); +async function syncFiles(info) { + const { + protocol, host, body, pid, domainId, docId, + } = info; + const judge = body.judgeInfo; + const r = await superagent.post(`${protocol}://${host === 'loj.ac' ? 'api.loj.ac.cn' : host}/api/problem/downloadProblemFiles`) + .send({ + problemId: +pid, + type: 'TestData', + filenameList: body.testData.map((node) => node.filename), + }); + const urls = {}; + if (r.body.error) return; + for (const t of r.body.downloadInfo) urls[t.filename] = t.downloadUrl; + for (const f of body.testData) { + const p = new PassThrough(); + superagent.get(urls[f.filename]).pipe(p); + // eslint-disable-next-line no-await-in-loop + await problem.addTestdata(domainId, docId, f.filename, p); + } + if (judge) { + const config = { + time: `${judge.timeLimit}ms`, + memory: `${judge.memoryLimit}m`, + // TODO other config + }; + await problem.addTestdata(domainId, docId, 'config.yaml', Buffer.from(yaml.dump(config))); + } + const a = await superagent.post(`${protocol}://${host === 'loj.ac' ? 'api.loj.ac.cn' : host}/api/problem/downloadProblemFiles`) + .send({ + problemId: +pid, + type: 'AdditionalFile', + filenameList: body.additionalFiles.map((node) => node.filename), + }); + const aurls = {}; + if (a.body.error) return; + for (const t of a.body.downloadInfo) aurls[t.filename] = t.downloadUrl; + for (const f of body.additionalFiles) { + const p = new PassThrough(); + superagent.get(aurls[f.filename]).pipe(p); + // eslint-disable-next-line no-await-in-loop + await problem.addAdditionalFile(domainId, docId, f.filename, p); + } +} +TaskModel.Worker.addHandler('import.syzoj', syncFiles); + class ProblemImportSYZOJHandler extends Handler { async get() { this.response.template = 'problem_import_syzoj.html'; @@ -174,49 +221,11 @@ class ProblemImportSYZOJHandler extends Handler { const docId = await problem.add( domainId, target, title, JSON.stringify(content), this.user._id, tags || [], hidden, ); - const syncFiles = async () => { - const judge = result.body.judgeInfo; - const r = await superagent.post(`${protocol}://${host === 'loj.ac' ? 'api.loj.ac.cn' : host}/api/problem/downloadProblemFiles`) - .send({ - problemId: +pid, - type: 'TestData', - filenameList: result.body.testData.map((node) => node.filename), - }); - const urls = {}; - if (r.body.error) return; - for (const t of r.body.downloadInfo) urls[t.filename] = t.downloadUrl; - for (const f of result.body.testData) { - const p = new PassThrough(); - superagent.get(urls[f.filename]).pipe(p); - // eslint-disable-next-line no-await-in-loop - await problem.addTestdata(domainId, docId, f.filename, p); - } - if (judge) { - const config = { - time: `${judge.timeLimit}ms`, - memory: `${judge.memoryLimit}m`, - // TODO other config - }; - await problem.addTestdata(domainId, docId, 'config.yaml', Buffer.from(yaml.dump(config))); - } - const a = await superagent.post(`${protocol}://${host === 'loj.ac' ? 'api.loj.ac.cn' : host}/api/problem/downloadProblemFiles`) - .send({ - problemId: +pid, - type: 'AdditionalFile', - filenameList: result.body.additionalFiles.map((node) => node.filename), - }); - const aurls = {}; - if (a.body.error) return; - for (const t of a.body.downloadInfo) aurls[t.filename] = t.downloadUrl; - for (const f of result.body.additionalFiles) { - const p = new PassThrough(); - superagent.get(aurls[f.filename]).pipe(p); - // eslint-disable-next-line no-await-in-loop - await problem.addAdditionalFile(domainId, docId, f.filename, p); - } + const payload = { + protocol, host, pid, domainId, docId, body: result.body, }; - if (wait) await syncFiles(); - else syncFiles().catch(logger.error); + if (wait) await syncFiles(payload); + else await TaskModel.add({ ...payload, type: 'schedule', subType: 'import.syzoj' }); return docId; } @@ -252,18 +261,11 @@ class ProblemImportSYZOJHandler extends Handler { assert(url.match(RE_SYZOJ), new ValidationError('url')); if (!url.endsWith('/')) url += '/'; const [, protocol, host, n, pid] = RE_SYZOJ.exec(url); - if (n === 'p') { - const docId = await this.v3( - domainId, targetPid, hidden, - protocol, host, pid, false, - ); - this.response.body = { pid: targetPid || docId }; - this.response.redirect = this.url('problem_files', { pid: targetPid || docId }); - } else { - const docId = await this.v2(domainId, targetPid, hidden, url); - this.response.body = { pid: targetPid || docId }; - this.response.redirect = this.url('problem_files', { pid: targetPid || docId }); - } + const docId = n === 'p' + ? await this.v3(domainId, targetPid, hidden, protocol, host, pid, false) + : await this.v2(domainId, targetPid, hidden, url); + this.response.body = { pid: targetPid || docId }; + this.response.redirect = this.url('problem_detail', { pid: targetPid || docId }); } } } diff --git a/packages/hydrooj/src/model/task.ts b/packages/hydrooj/src/model/task.ts index 696f6e355..24bffcb8a 100644 --- a/packages/hydrooj/src/model/task.ts +++ b/packages/hydrooj/src/model/task.ts @@ -1,12 +1,89 @@ import moment from 'moment-timezone'; import { FilterQuery, ObjectID } from 'mongodb'; -import { Task } from '../interface'; +import { BaseService, Task } from '../interface'; import { Logger } from '../logger'; import db from '../service/db'; const logger = new Logger('model/task'); const coll = db.collection('task'); +async function getFirst(query: FilterQuery) { + const q = { ...query }; + q.executeAfter = q.executeAfter || { $lt: new Date() }; + const res = await coll.findOneAndDelete(q, { sort: { priority: -1 } }); + if (res.value) { + logger.debug('%o', res.value); + if (res.value.interval) { + const executeAfter = moment(res.value.executeAfter).add(...res.value.interval).toDate(); + await coll.insertOne({ ...res.value, executeAfter }); + } + return res.value; + } + return null; +} + +class Consumer { + consuming: boolean; + running?: any; + filter: any; + func: Function; + timeout: NodeJS.Timeout; + + constructor(filter: any, func: Function) { + this.consuming = true; + this.filter = filter; + this.func = func; + this.get = this.get.bind(this); + this.get(); + } + + async get() { + if (this.running || !this.consuming) return; + try { + const res = await getFirst(this.filter); + if (res) { + this.running = res; + await this.func(res); + this.running = null; + } + } catch (e) { + logger.error(e); + this.destory(); + } + this.timeout = setTimeout(this.get, 100); + } + + async destory() { + this.consuming = false; + clearTimeout(this.timeout); + } +} + +class WorkerService implements BaseService { + private handlers: Record = {}; + public consumer: Consumer; + public started = false; + + public start() { + this.consumer = new Consumer( + { type: 'schedule', subType: { $in: Object.keys(this.handlers) } }, + (doc) => { + logger.debug('Worker task: %o', doc); + return this.handlers[doc.subType](doc); + }, + ); + this.started = true; + } + + public addHandler(type: string, handler: Function) { + this.handlers[type] = handler; + this.consumer.filter = { type: 'schedule', subType: { $in: Object.keys(this.handlers) } }; + } +} + +const Worker = new WorkerService(); +Worker.start(); + class TaskModel { static async add(task: Partial & { type: string }) { const t: Task = { @@ -32,21 +109,7 @@ class TaskModel { return coll.deleteOne({ _id }); } - static async getFirst(query: FilterQuery) { - const q = { ...query }; - q.executeAfter = q.executeAfter || { $lt: new Date() }; - const res = await coll.findOneAndDelete(q, { sort: { priority: -1 } }); - if (res.value) { - logger.debug('%o', res.value); - if (res.value.interval) { - await coll.insertOne({ - ...res.value, executeAfter: moment().add(...res.value.interval).toDate(), - }); - } - return res.value; - } - return null; - } + static getFirst = getFirst; static async getDelay(query?: FilterQuery) { const now = new Date(); @@ -56,52 +119,12 @@ class TaskModel { } static async consume(query: any, cb: Function) { - let isRunning = false; - const interval = setInterval(async () => { - if (isRunning) return; - isRunning = true; - const res = await TaskModel.getFirst(query); - if (res) { - try { - await cb(res); - } catch (e) { - clearInterval(interval); - } - } - isRunning = false; - }, 100); + return new Consumer(query, cb); } - Consumer = class Consumer { - consuming: boolean; - running?: Promise; - interval: NodeJS.Timeout; - filter: any; - func: Function; - - constructor(filter: any, func: Function) { - this.consuming = true; - this.filter = filter; - this.func = func; - this.interval = setInterval(this.consume.bind(this), 100); - } - - async consume() { - if (this.running || !this.consuming) return; - const res = await TaskModel.getFirst(this.filter); - if (res) { - this.running = this.func(res); - if (this.running instanceof Promise) await this.running; - this.running = null; - } - } - - async destory() { - this.consuming = false; - clearInterval(this.interval); - if (this.running) await this.running; - } - }; + static Consumer = Consumer; + static WorkerService = WorkerService; + static Worker = Worker; } export default TaskModel; diff --git a/packages/hydrooj/src/ui.ts b/packages/hydrooj/src/ui.ts index 7f2b6c494..0a6be6ee5 100644 --- a/packages/hydrooj/src/ui.ts +++ b/packages/hydrooj/src/ui.ts @@ -1,12 +1,8 @@ /* eslint-disable @typescript-eslint/no-shadow */ import cluster from 'cluster'; import { argv } from 'yargs'; -import BottomBar from 'inquirer/lib/ui/bottom-bar'; -import { Logger } from './logger'; import * as bus from './service/bus'; -const logger = new Logger('ui'); -const useTerminal = cluster.isMaster && process.stdout.isTTY; const disabledTerminal = argv.legacy || argv._.length || process.env.NODE_ENV === 'test'; export namespace Progress { @@ -43,45 +39,7 @@ async function terminate() { process.exit(hasError ? 1 : 0); } process.on('SIGINT', terminate); - -if (useTerminal && !disabledTerminal) { - let current = 0; - const history = ['']; - let input = ''; - const ui = new BottomBar({ bottomBar: '>' }); - process.stdin.on('data', (data) => { - // Control Seq? - const seq = data.toString('hex'); - if (seq === '7f') input = input.substr(0, input.length - 1); - else if (seq === '1b5b41') { - // Arrow up - current--; - input = history[current]; - } else if (seq === '1b5b42') { - // Arrow down - current++; - input = history[current]; - } else if (seq === '0d') { - // Enter - history.push(input); - current = history.length; - if (input[0] === '@') { - for (const i in cluster.workers) { - cluster.workers[i].send({ event: 'message/run', payload: [input.substr(1, input.length - 1)] }); - break; - } - } else bus.parallel('message/run', input); - input = ''; - } else input += data.toString(); - ui.updateBottomBar(`>${input}`); - }); - bus.on('message/log', (message) => { - ui.log.write(message); - }); - if (process.env.SSH_CLIENT || process.env.SSH_TTY || process.env.SSH_CONNECTION) { - logger.warn('Running over ssh detected. Add a --legacy when starting if GUI mode doesn\'t work properly.'); - } -} else if (cluster.isMaster) { +if (cluster.isMaster) { if (!disabledTerminal) console.log('Not running in a terminal environment. Interactive mode disabled.'); bus.on('message/log', (message) => { process.stdout.write(`${message}\n`); diff --git a/packages/ui-default/package.json b/packages/ui-default/package.json index a2af14586..0b23501be 100644 --- a/packages/ui-default/package.json +++ b/packages/ui-default/package.json @@ -8,7 +8,7 @@ "devDependencies": { "@babel/cli": "^7.13.14", "@babel/core": "^7.14.0", - "@babel/eslint-parser": "7.13.14", + "@babel/eslint-parser": "7.14.2", "@babel/plugin-proposal-class-properties": "^7.13.0", "@babel/plugin-proposal-export-namespace-from": "^7.12.13", "@babel/plugin-proposal-function-sent": "^7.12.13", @@ -44,13 +44,13 @@ "eslint-config-airbnb": "^18.2.1", "eslint-import-resolver-webpack": "^0.13.0", "eslint-plugin-babel": "^5.3.1", - "eslint-plugin-import": "^2.22.1", + "eslint-plugin-import": "^2.23.2", "eslint-plugin-jsx-a11y": "^6.4.1", "eslint-plugin-react": "^7.23.1", "fancy-log": "^1.3.3", "file-loader": "^6.2.0", "friendly-errors-webpack-plugin": "^1.7.0", - "fs-extra": "^9.1.0", + "fs-extra": "^10.0.0", "gulp": "^4.0.2", "gulp-cli": "^2.3.0", "gulp-iconfont": "^11.0.0", @@ -64,7 +64,7 @@ "jquery.transit": "^0.9.12", "mini-css-extract-plugin": "^1.6.0", "moment": "^2.29.1", - "monaco-editor": "^0.23.0", + "monaco-editor": "^0.24.0", "monaco-editor-webpack-plugin": "^3.0.1", "normalize.css": "^8.0.1", "nprogress": "^0.2.0", @@ -100,14 +100,14 @@ "vinyl-buffer": "^1.0.1", "wastyle": "^0.0.5", "webpack": "^4.44.2", - "webpack-bundle-analyzer": "^4.4.0", + "webpack-bundle-analyzer": "^4.4.2", "webpack-dev-server": "^3.11.2", "webpackbar": "^5.0.0-3" }, "dependencies": { "98.css": "^0.1.16", "js-yaml": "^4.0.0", - "katex": "^0.13.9", + "katex": "^0.13.11", "lodash": "^4.17.21", "markdown-it": "^12.0.6", "markdown-it-anchor": "^7.1.0",