From 636446c43a8d23393ebba3ea41dad34ec5b22ca2 Mon Sep 17 00:00:00 2001 From: undefined Date: Sun, 28 Aug 2022 04:16:06 +0800 Subject: [PATCH] judge: refactor: use common JudgeTask interface --- packages/hydrojudge/package.json | 2 +- packages/hydrojudge/src/checkers.ts | 9 +- packages/hydrojudge/src/compile.ts | 3 +- packages/hydrojudge/src/daemon.ts | 4 +- packages/hydrojudge/src/error.ts | 7 +- packages/hydrojudge/src/hosts/builtin.ts | 82 +++++++ packages/hydrojudge/src/hosts/hydro.ts | 175 +++------------ packages/hydrojudge/src/hosts/index.ts | 3 - packages/hydrojudge/src/hosts/vj4.ts | 9 - packages/hydrojudge/src/interface.ts | 23 +- packages/hydrojudge/src/judge/default.ts | 14 +- packages/hydrojudge/src/judge/index.ts | 2 +- packages/hydrojudge/src/judge/interactive.ts | 9 +- packages/hydrojudge/src/judge/interface.ts | 22 +- packages/hydrojudge/src/judge/run.ts | 4 +- .../hydrojudge/src/judge/submit_answer.ts | 10 +- packages/hydrojudge/src/sandbox.ts | 2 + packages/hydrojudge/src/service.ts | 205 +----------------- packages/hydrojudge/src/sysinfo.ts | 80 ------- packages/hydrojudge/src/task.ts | 132 +++++++++++ 20 files changed, 291 insertions(+), 506 deletions(-) create mode 100644 packages/hydrojudge/src/hosts/builtin.ts delete mode 100644 packages/hydrojudge/src/hosts/index.ts delete mode 100644 packages/hydrojudge/src/hosts/vj4.ts delete mode 100644 packages/hydrojudge/src/sysinfo.ts create mode 100644 packages/hydrojudge/src/task.ts diff --git a/packages/hydrojudge/package.json b/packages/hydrojudge/package.json index ef907380d..f034e0776 100644 --- a/packages/hydrojudge/package.json +++ b/packages/hydrojudge/package.json @@ -1,7 +1,7 @@ { "name": "@hydrooj/hydrojudge", "bin": "bin/hydrojudge.js", - "version": "2.22.13", + "version": "3.0.0", "main": "package.json", "author": "undefined ", "repository": "https://github.com/hydro-dev/Hydro.git", diff --git a/packages/hydrojudge/src/checkers.ts b/packages/hydrojudge/src/checkers.ts index 2f39675d4..91bca0cd4 100644 --- a/packages/hydrojudge/src/checkers.ts +++ b/packages/hydrojudge/src/checkers.ts @@ -1,11 +1,16 @@ /* eslint-disable no-template-curly-in-string */ import { STATUS } from '@hydrooj/utils/lib/status'; import { SystemError } from './error'; -import { CheckConfig, CheckResult } from './interface'; +import { CheckConfig } from './interface'; import { run } from './sandbox'; import { parse } from './testlib'; -type Checker = (config: CheckConfig) => Promise; +type Checker = (config: CheckConfig) => Promise<{ + status: number, + score: number, + message: string, + code?: number, +}>; const checkers: Record = { async default(config) { diff --git a/packages/hydrojudge/src/compile.ts b/packages/hydrojudge/src/compile.ts index 76c170c5c..679461917 100644 --- a/packages/hydrojudge/src/compile.ts +++ b/packages/hydrojudge/src/compile.ts @@ -2,8 +2,7 @@ import { LangConfig } from '@hydrooj/utils/lib/lang'; import { STATUS } from '@hydrooj/utils/lib/status'; import { CompileError } from './error'; import { Execute } from './interface'; -import { del, run } from './sandbox'; -import { CopyInFile } from './sandbox/interface'; +import { CopyInFile, del, run } from './sandbox'; import { compilerText } from './utils'; export = async function compile( diff --git a/packages/hydrojudge/src/daemon.ts b/packages/hydrojudge/src/daemon.ts index 02cfe91ad..180aeab98 100644 --- a/packages/hydrojudge/src/daemon.ts +++ b/packages/hydrojudge/src/daemon.ts @@ -16,7 +16,7 @@ import './utils'; import { getConfig } from './config'; -import * as Session from './hosts/index'; +import HydroHost from './hosts/hydro'; import log from './log'; import { Queue } from './utils'; @@ -65,7 +65,7 @@ async function daemon() { worker(queue).catch((e) => log.error(e)); for (const i in _hosts) { _hosts[i].host = _hosts[i].host || i; - hosts[i] = new Session[_hosts[i].type || 'hydro'](_hosts[i]); + hosts[i] = new HydroHost(_hosts[i]); await hosts[i].init(); } for (const i in hosts) hosts[i].consume(queue); diff --git a/packages/hydrojudge/src/error.ts b/packages/hydrojudge/src/error.ts index ff79b9a0b..1f1385749 100644 --- a/packages/hydrojudge/src/error.ts +++ b/packages/hydrojudge/src/error.ts @@ -1,5 +1,10 @@ import { STATUS_TEXTS } from '@hydrooj/utils/lib/status'; -import { CompileErrorInfo } from './interface'; + +interface CompileErrorInfo { + stdout?: string, + stderr?: string, + status?: number +} export class CompileError extends Error { stdout: string; diff --git a/packages/hydrojudge/src/hosts/builtin.ts b/packages/hydrojudge/src/hosts/builtin.ts new file mode 100644 index 000000000..96e15f1e0 --- /dev/null +++ b/packages/hydrojudge/src/hosts/builtin.ts @@ -0,0 +1,82 @@ +// Hydro Integration +/* eslint-disable no-await-in-loop */ +import path from 'path'; +import fs from 'fs-extra'; +import { ObjectID } from 'mongodb'; +import type { JudgeResultBody } from 'hydrooj'; +import { end, next } from 'hydrooj/src/handler/judge'; +import { Logger } from 'hydrooj/src/logger'; +import * as setting from 'hydrooj/src/model/setting'; +import * as system from 'hydrooj/src/model/system'; +import task from 'hydrooj/src/model/task'; +import * as monitor from 'hydrooj/src/service/monitor'; +import storage from 'hydrooj/src/service/storage'; +import { processTestdata } from '../cases'; +import { getConfig } from '../config'; +import { FormatError, SystemError } from '../error'; +import { Context } from '../judge/interface'; +import { JudgeTask } from '../task'; + +const logger = new Logger('judge'); + +const session = { + config: { detail: system.get('hydrojudge.detail') }, + async fetchCodeFile(name: string) { + const target = path.join('/tmp/hydro/judge', name.replace(/\//g, '_')); + await storage.get(`submission/${name}`, target); + return target; + }, + getNext(t: Context) { + return (data: Partial) => { + logger.debug('Next: %d %o', data); + data.rid = new ObjectID(t.rid); + if (data.case) data.case.message ||= ''; + next(data); + }; + }, + getEnd(t: Context) { + return (data: Partial) => { + data.key = 'end'; + data.rid = new ObjectID(t.rid); + logger.info('End: status=%d score=%d time=%dms memory=%dkb', data.status, data.score, data.time, data.memory); + end(data); + }; + }, + getLang(lang: string, doThrow = true) { + if (setting.langs[lang]) return setting.langs[lang]; + if (lang === 'cpp' && setting.langs['cc']) return setting.langs['cc']; + if (doThrow) throw new SystemError('Unsupported language {0}.', [lang]); + return null; + }, + async cacheOpen(source: string, files: any[]) { + const filePath = path.join(getConfig('cache_dir'), source); + await fs.ensureDir(filePath); + if (!files?.length) throw new FormatError('Problem data not found.'); + let etags: Record = {}; + try { + etags = JSON.parse(fs.readFileSync(path.join(filePath, 'etags')).toString()); + } catch (e) { /* ignore */ } + const version = {}; + const filenames = new Set(); + for (const file of files) { + filenames.add(file.name); + version[file.name] = file.etag + file.lastModified; + if (etags[file.name] !== file.etag + file.lastModified) { + await storage.get(`problem/${source}/testdata/${file.name}`, path.join(filePath, file.name)); + } + } + for (const name in etags) { + if (!filenames.has(name) && fs.existsSync(path.join(filePath, name))) await fs.remove(path.join(filePath, name)); + } + fs.writeFileSync(path.join(filePath, 'etags'), JSON.stringify(version)); + fs.writeFileSync(path.join(filePath, 'lastUsage'), Date.now().toString()); + await processTestdata(filePath); + return filePath; + }, +}; + +export async function postInit() { + if (system.get('hydrojudge.disable')) return; + task.consume({ type: 'judge' }, (t) => (new JudgeTask(session, t)).handle().catch(logger.error)); + task.consume({ type: 'judge', priority: { $gt: -50 } }, (t) => (new JudgeTask(session, t)).handle().catch(logger.error)); +} diff --git a/packages/hydrojudge/src/hosts/hydro.ts b/packages/hydrojudge/src/hosts/hydro.ts index 161f243ba..d5fdfac2a 100644 --- a/packages/hydrojudge/src/hosts/hydro.ts +++ b/packages/hydrojudge/src/hosts/hydro.ts @@ -1,162 +1,24 @@ /* eslint-disable no-await-in-loop */ import path from 'path'; import axios from 'axios'; -import cac from 'cac'; +import { ObjectID } from 'bson'; import fs from 'fs-extra'; -import { noop } from 'lodash'; -import { ObjectID } from 'mongodb'; import PQueue from 'p-queue'; import WebSocket from 'ws'; import { LangConfig } from '@hydrooj/utils/lib/lang'; -import { STATUS } from '@hydrooj/utils/lib/status'; +import * as sysinfo from '@hydrooj/utils/lib/sysinfo'; import type { JudgeResultBody } from 'hydrooj'; -import readCases, { processTestdata } from '../cases'; +import { processTestdata } from '../cases'; import { getConfig } from '../config'; -import { CompileError, FormatError, SystemError } from '../error'; -import judge from '../judge'; +import { FormatError, SystemError } from '../error'; import log from '../log'; -import type { CopyInFile } from '../sandbox/interface'; -import * as sysinfo from '../sysinfo'; -import * as tmpfs from '../tmpfs'; -import { - compilerText, Lock, md5, Queue, -} from '../utils'; - -const argv = cac().parse(); +import { JudgeTask } from '../task'; +import { Lock, Queue } from '../utils'; function removeNixPath(text: string) { return text.replace(/\/nix\/store\/[a-z0-9]{32}-/g, '/nix/'); } -class JudgeTask { - stat: Record; - session: any; - host: string; - request: any; - ws: WebSocket; - source: string; - rid: string; - lang: string; - code: CopyInFile; - tmpdir: string; - input?: string; - clean: (() => Promise)[]; - data: any[]; - folder: string; - config: any; - env: Record; - getLang: (name: string) => LangConfig; - - constructor(session: Hydro, request, ws: WebSocket) { - this.stat = {}; - this.stat.receive = new Date(); - this.session = session; - this.host = session.config.host; - this.request = request; - this.ws = ws; - this.getLang = session.getLang; - } - - async handle(startPromise = Promise.resolve()) { - this.next = this.next.bind(this); - this.end = this.end.bind(this); - this.stat.handle = new Date(); - this.rid = this.request.rid; - this.lang = this.request.lang; - this.code = { content: this.request.code }; - this.config = this.request.config; - this.input = this.request.input; - this.data = this.request.data; - this.source = this.request.source; - this.tmpdir = path.resolve(getConfig('tmp_dir'), this.host, this.rid); - this.clean = []; - let tid = this.request.contest?.toString() || ''; - if (tid === '000000000000000000000000') tid = ''; - this.env = { - HYDRO_DOMAIN: this.request.domainId.toString(), - HYDRO_RECORD: this.rid, - HYDRO_LANG: this.lang, - HYDRO_USER: this.request.uid.toString(), - HYDRO_CONTEST: tid, - }; - await Lock.acquire(`${this.host}/${this.source}/${this.rid}`); - fs.ensureDirSync(this.tmpdir); - tmpfs.mount(this.tmpdir, getConfig('tmpfs_size')); - log.info('Submission: %s/%s/%s', this.host, this.source, this.rid); - try { - await this.doSubmission(startPromise); - } catch (e) { - if (e instanceof CompileError) { - this.next({ compilerText: compilerText(e.stdout, e.stderr) }); - this.end({ - status: STATUS.STATUS_COMPILE_ERROR, score: 0, time: 0, memory: 0, - }); - } else if (e instanceof FormatError) { - this.next({ message: 'Testdata configuration incorrect.' }); - this.next({ message: { message: e.message, params: e.params } }); - this.end({ - status: STATUS.STATUS_FORMAT_ERROR, score: 0, time: 0, memory: 0, - }); - } else { - log.error(e); - this.next({ message: { message: e.message, params: e.params || [], ...argv.options.debug ? { stack: e.stack } : {} } }); - this.end({ - status: STATUS.STATUS_SYSTEM_ERROR, score: 0, time: 0, memory: 0, - }); - } - } finally { - Lock.release(`${this.host}/${this.source}/${this.rid}`); - for (const clean of this.clean) await clean()?.catch(noop); - tmpfs.umount(this.tmpdir); - fs.removeSync(this.tmpdir); - } - } - - async doSubmission(startPromise = Promise.resolve()) { - this.stat.cache_start = new Date(); - this.folder = await this.session.cacheOpen(this.source, this.data, this.next); - if ((this.code as any).content.startsWith('@@hydro_submission_file@@')) { - const id = (this.code as any).content.split('@@hydro_submission_file@@')[1]?.split('#')?.[0]; - const target = await this.session.fetchCodeFile(id); - this.code = { src: target }; - this.clean.push(() => fs.remove(target)); - } - this.stat.read_cases = new Date(); - this.config = await readCases( - this.folder, - { - detail: this.session.config.detail, - isSelfSubmission: this.config.problemOwner === this.request.uid, - ...this.config, - }, - { next: this.next, key: md5(`${this.source}/${getConfig('secret')}`) }, - ); - this.stat.judge = new Date(); - const type = typeof this.input === 'string' ? 'run' : this.config.type || 'default'; - if (!judge[type]) throw new FormatError('Unrecognized problemType: {0}', [type]); - await judge[type].judge(this, startPromise); - } - - next(data: Partial) { - log.debug('Next: %d %o', data); - data.key = 'next'; - data.rid = new ObjectID(this.rid); - if (data.case) data.case.message ||= ''; - if (typeof data.message === 'string') data.message = removeNixPath(data.message); - if (typeof data.compilerText === 'string') data.compilerText = removeNixPath(data.compilerText); - this.ws.send(JSON.stringify(data)); - } - - end(data: Partial) { - log.info('End: %o', data); - data.key = 'end'; - data.rid = this.request.rid; - if (typeof data.message === 'string') data.message = removeNixPath(data.message); - if (typeof data.compilerText === 'string') data.compilerText = removeNixPath(data.compilerText); - this.ws.send(JSON.stringify(data)); - } -} - export default class Hydro { config: any; axios: any; @@ -265,6 +127,29 @@ export default class Hydro { return null; } + send(rid: string, key: 'next' | 'end', data: Partial) { + data.rid = new ObjectID(rid); + data.key = key; + if (data.case) data.case.message ||= ''; + if (typeof data.message === 'string') data.message = removeNixPath(data.message); + if (typeof data.compilerText === 'string') data.compilerText = removeNixPath(data.compilerText); + this.ws.send(JSON.stringify(data)); + } + + getNext(t: JudgeTask) { + return (data: Partial) => { + log.debug('Next: %d %o', data); + this.send(t.request.rid, 'next', data); + }; + } + + getEnd(t: JudgeTask) { + return (data: Partial) => { + log.info('End: %o', data); + this.send(t.request.rid, 'end', data); + }; + } + async consume(queue: Queue) { log.info('正在连接 %sjudge/conn', this.config.server_url); this.ws = new WebSocket(`${this.config.server_url.replace(/^http/i, 'ws')}judge/conn`, { @@ -280,7 +165,7 @@ export default class Hydro { this.ws.on('message', (data) => { const request = JSON.parse(data.toString()); if (request.language) this.language = request.language; - if (request.task) queue.push(new JudgeTask(this, request.task, this.ws)); + if (request.task) queue.push(new JudgeTask(this, request.task)); }); this.ws.on('close', (data, reason) => { log.warn(`[${this.config.host}] Websocket 断开:`, data, reason.toString()); diff --git a/packages/hydrojudge/src/hosts/index.ts b/packages/hydrojudge/src/hosts/index.ts deleted file mode 100644 index f8af9b618..000000000 --- a/packages/hydrojudge/src/hosts/index.ts +++ /dev/null @@ -1,3 +0,0 @@ -export { default as vj4 } from './vj4'; -// eslint-disable-next-line no-restricted-exports -export { default as hydro, default } from './hydro'; diff --git a/packages/hydrojudge/src/hosts/vj4.ts b/packages/hydrojudge/src/hosts/vj4.ts deleted file mode 100644 index 07590d907..000000000 --- a/packages/hydrojudge/src/hosts/vj4.ts +++ /dev/null @@ -1,9 +0,0 @@ -export default class VJ4 { - constructor() { - console.log('新版本不再支持 VJ4。请降级至 hydrojudge@2.5.21 或迁移到 Hydro。'); - console.log('This version of HydroJudge nolonger supports VJ4. Please downgrade to'); - console.log('hydrojudge@2.5.21 or migrate to Hydro.'); - console.log('详细文档： https://hydro.js.org/ '); - process.exit(1); - } -} diff --git a/packages/hydrojudge/src/interface.ts b/packages/hydrojudge/src/interface.ts index e61f2f148..1014c4c15 100644 --- a/packages/hydrojudge/src/interface.ts +++ b/packages/hydrojudge/src/interface.ts @@ -1,16 +1,10 @@ -import { CopyInFile } from './sandbox/interface'; +import { CopyInFile } from './sandbox'; export interface Execute { - execute: string, - clean: Function, - copyIn: Record, - time?: number, -} - -export interface CompileErrorInfo { - stdout?: string, - stderr?: string, - status?: number + execute: string; + clean: () => Promise; + copyIn: Record; + time?: number; } export interface CheckConfig { @@ -25,10 +19,3 @@ export interface CheckConfig { detail: boolean; env?: Record; } - -export interface CheckResult { - status: number, - score: number, - message: string, - code?: number, -} diff --git a/packages/hydrojudge/src/judge/default.ts b/packages/hydrojudge/src/judge/default.ts index 4316bc2d2..c6e5a9198 100644 --- a/packages/hydrojudge/src/judge/default.ts +++ b/packages/hydrojudge/src/judge/default.ts @@ -5,8 +5,7 @@ import compile from '../compile'; import { getConfig } from '../config'; import { CompileError, FormatError } from '../error'; import { Logger } from '../log'; -import { del, run } from '../sandbox'; -import { CmdFile } from '../sandbox/interface'; +import { CmdFile, del, run } from '../sandbox'; import signals from '../signals'; import { NormalizedCase, NormalizedSubtask, parseFilename } from '../utils'; import { Context, ContextSubTask } from './interface'; @@ -113,7 +112,7 @@ function judgeCase(c: NormalizedCase, sid: string) { }); if (ctx.request.rejudged) return; // Skip analysis for rejudged submissions if ([STATUS.STATUS_WRONG_ANSWER, STATUS.STATUS_RUNTIME_ERROR].includes(status)) { - const langConfig = ctx.getLang(ctx.lang); + const langConfig = ctx.session.getLang(ctx.lang); if (langConfig.analysis && !ctx.analysis) { ctx.analysis = true; try { @@ -165,9 +164,9 @@ function judgeSubtask(subtask: NormalizedSubtask, sid: string) { }; } -export const judge = async (ctx: Context, startPromise = Promise.resolve()) => { +export const judge = async (ctx: Context) => { if (!ctx.config.subtasks.length) throw new FormatError('Problem data not found.'); - startPromise.then(() => ctx.next({ status: STATUS.STATUS_COMPILING })); + ctx.next({ status: STATUS.STATUS_COMPILING }); if (ctx.config.template && 'content' in ctx.code) { if (ctx.config.template[ctx.lang]) { const tpl = ctx.config.template[ctx.lang]; @@ -176,14 +175,14 @@ export const judge = async (ctx: Context, startPromise = Promise.resolve()) => { } [ctx.execute, ctx.checker] = await Promise.all([ compile( - ctx.getLang(ctx.lang), ctx.code, + ctx.session.getLang(ctx.lang), ctx.code, Object.fromEntries( (ctx.config.user_extra_files || []).map((i) => [i.split('/').pop(), { src: i }]), ), ctx.next, ), compileChecker( - ctx.getLang, + ctx.session.getLang, ctx.config.checker_type, ctx.config.checker, { @@ -195,7 +194,6 @@ export const judge = async (ctx: Context, startPromise = Promise.resolve()) => { ), ]); ctx.clean.push(ctx.execute.clean, ctx.checker.clean); - await startPromise; ctx.next({ status: STATUS.STATUS_JUDGING, progress: 0 }); const tasks = []; ctx.total_status = 0; diff --git a/packages/hydrojudge/src/judge/index.ts b/packages/hydrojudge/src/judge/index.ts index fc4cc2b4a..ee431ccc9 100644 --- a/packages/hydrojudge/src/judge/index.ts +++ b/packages/hydrojudge/src/judge/index.ts @@ -7,4 +7,4 @@ import * as submit_answer from './submit_answer'; export = { default: def, interactive, run, submit_answer, objective, -} as Record): Promise }>; +} as Record }>; diff --git a/packages/hydrojudge/src/judge/interactive.ts b/packages/hydrojudge/src/judge/interactive.ts index a2201a170..f2c71fc06 100644 --- a/packages/hydrojudge/src/judge/interactive.ts +++ b/packages/hydrojudge/src/judge/interactive.ts @@ -96,15 +96,15 @@ function judgeSubtask(subtask: NormalizedSubtask) { }; } -export const judge = async (ctx: Context, startPromise = Promise.resolve()) => { - startPromise.then(() => ctx.next({ status: STATUS.STATUS_COMPILING })); +export const judge = async (ctx: Context) => { + ctx.next({ status: STATUS.STATUS_COMPILING }); [ctx.executeUser, ctx.executeInteractor] = await Promise.all([ (() => { const copyIn = {}; for (const file of ctx.config.user_extra_files) { copyIn[parseFilename(file)] = { src: file }; } - return compile(ctx.getLang(ctx.lang), ctx.code, copyIn, ctx.next); + return compile(ctx.session.getLang(ctx.lang), ctx.code, copyIn, ctx.next); })(), (() => { const copyIn = { @@ -115,14 +115,13 @@ export const judge = async (ctx: Context, startPromise = Promise.resolve()) => { copyIn[parseFilename(file)] = { src: file }; } return compile( - ctx.getLang(parseFilename(ctx.config.interactor).split('.')[1].replace('@', '.')), + ctx.session.getLang(parseFilename(ctx.config.interactor).split('.')[1].replace('@', '.')), { src: ctx.config.interactor }, copyIn, ); })(), ]); ctx.clean.push(ctx.executeUser.clean, ctx.executeInteractor.clean); - await startPromise; ctx.next({ status: STATUS.STATUS_JUDGING, progress: 0 }); const tasks = []; ctx.total_status = ctx.total_score = ctx.total_memory_usage_kb = ctx.total_time_usage_ms = 0; diff --git a/packages/hydrojudge/src/judge/interface.ts b/packages/hydrojudge/src/judge/interface.ts index b7a9cab9b..f48aa2321 100644 --- a/packages/hydrojudge/src/judge/interface.ts +++ b/packages/hydrojudge/src/judge/interface.ts @@ -1,27 +1,9 @@ import PQueue from 'p-queue'; import type { NormalizedSubtask } from '@hydrooj/utils/lib/common'; -import type { LangConfig } from '@hydrooj/utils/lib/lang'; -import type { JudgeResultBody } from 'hydrooj'; import { Execute } from '../interface'; -import { CopyInFile } from '../sandbox/interface'; +import { JudgeTask } from '../task'; -export type Context = JudgeTaskInterface & RuntimeContext; - -export interface JudgeTaskInterface { - next(data: Partial, id?: number): void; - end(data: Partial): void; - getLang: (name: string) => LangConfig; - - stat: Record; - lang: string; - code: CopyInFile; - tmpdir: string; - input?: string; - clean: Function[]; - config: Config; - env?: Record; - request: Record; -} +export type Context = JudgeTask & RuntimeContext; export interface RuntimeContext { total_score?: number; diff --git a/packages/hydrojudge/src/judge/run.ts b/packages/hydrojudge/src/judge/run.ts index c87503ec8..54087444f 100644 --- a/packages/hydrojudge/src/judge/run.ts +++ b/packages/hydrojudge/src/judge/run.ts @@ -24,7 +24,7 @@ export const judge = async (ctx: Context) => { ctx.next({ status: STATUS.STATUS_COMPILING }); try { ctx.execute = await compile( - ctx.getLang(ctx.lang), ctx.code, + ctx.session.getLang(ctx.lang), ctx.code, Object.fromEntries( (ctx.config.user_extra_files || []).map((i) => [i.split('/').pop(), { src: i }]), ), @@ -97,7 +97,7 @@ export const judge = async (ctx: Context) => { }, }); if ([STATUS.STATUS_WRONG_ANSWER, STATUS.STATUS_RUNTIME_ERROR].includes(status)) { - const langConfig = ctx.getLang(ctx.lang); + const langConfig = ctx.session.getLang(ctx.lang); if (langConfig.analysis) { try { ctx.analysis = true; diff --git a/packages/hydrojudge/src/judge/submit_answer.ts b/packages/hydrojudge/src/judge/submit_answer.ts index a0ca623e9..6e273321e 100644 --- a/packages/hydrojudge/src/judge/submit_answer.ts +++ b/packages/hydrojudge/src/judge/submit_answer.ts @@ -22,6 +22,7 @@ function judgeCase(c: NormalizedCase, sid: string) { ) { ctx.next({ case: { + id: c.id, status: STATUS.STATUS_CANCELED, score: 0, time: 0, @@ -29,7 +30,7 @@ function judgeCase(c: NormalizedCase, sid: string) { message: '', }, addProgress: 100 / ctx.config.count, - }, c.id); + }); return; } const chars = /[a-zA-Z0-9_.-]/; @@ -119,23 +120,22 @@ function judgeSubtask(subtask: NormalizedSubtask, sid: string) { }; } -export const judge = async (ctx: Context, startPromise = Promise.resolve()) => { +export const judge = async (ctx: Context) => { if (!ctx.config.subtasks.length) throw new FormatError('Problem data not found.'); - startPromise.then(() => ctx.next({ status: STATUS.STATUS_COMPILING })); + ctx.next({ status: STATUS.STATUS_COMPILING }); ctx.checker = await (() => { const copyIn = { user_code: ctx.code }; for (const file of ctx.config.judge_extra_files) { copyIn[parseFilename(file)] = { src: file }; } return compileChecker( - ctx.getLang, + ctx.session.getLang, ctx.config.checker_type, ctx.config.checker, copyIn, ); })(); ctx.clean.push(ctx.checker.clean); - await startPromise; ctx.next({ status: STATUS.STATUS_JUDGING, progress: 0 }); const tasks = []; ctx.total_status = 0; diff --git a/packages/hydrojudge/src/sandbox.ts b/packages/hydrojudge/src/sandbox.ts index feb6a5a0c..d0eac3f44 100644 --- a/packages/hydrojudge/src/sandbox.ts +++ b/packages/hydrojudge/src/sandbox.ts @@ -198,3 +198,5 @@ const queue = new PQueue({ concurrency: getConfig('parallelism') }); export function runQueued(execute: string, params?: Parameter, priority = 0) { return queue.add(() => run(execute, params), { priority }); } + +export * from './sandbox/interface'; diff --git a/packages/hydrojudge/src/service.ts b/packages/hydrojudge/src/service.ts index 5611bd91d..286b3e724 100644 --- a/packages/hydrojudge/src/service.ts +++ b/packages/hydrojudge/src/service.ts @@ -1,22 +1,4 @@ -// Hydro Integration -/* eslint-disable no-await-in-loop */ -import 'hydrooj'; - -import path from 'path'; -import fs from 'fs-extra'; -import { noop } from 'lodash'; -import { ObjectID } from 'mongodb'; -import { STATUS } from '@hydrooj/utils/lib/status'; -import type { JudgeResultBody } from 'hydrooj'; -import { Logger } from 'hydrooj/src/logger'; -import * as monitor from 'hydrooj/src/service/monitor'; -import readCases, { processTestdata } from './cases'; -import { getConfig } from './config'; -import { CompileError, FormatError, SystemError } from './error'; -import { CopyInFile } from './sandbox/interface'; -import * as sysinfo from './sysinfo'; -import * as tmpfs from './tmpfs'; -import { compilerText, md5 } from './utils'; +import * as bus from 'hydrooj/src/service/bus'; declare module 'hydrooj/src/interface' { interface SystemKeys { @@ -33,187 +15,6 @@ declare module 'hydrooj/src/interface' { } } -const logger = new Logger('judge'); - -async function postInit() { - // Only start a single daemon - if (process.env.NODE_APP_INSTANCE !== '0') return; - const judge = require('./judge'); - - const { - task, system, setting, storage, - } = global.Hydro.model; - const _judge = global.Hydro.handler.judge as any; - - if (system.get('hydrojudge.disable')) return; - const info = await sysinfo.get(); - monitor.updateJudge(info); - setInterval(async () => { - const [mid, inf] = await sysinfo.update(); - monitor.updateJudge({ mid, ...inf }); - }, 1200000); - - async function fetchCodeFile(name: string) { - const target = path.join('/tmp/hydro/judge', name.replace(/\//g, '_')); - await storage.get(`submission/${name}`, target); - return target; - } - - async function cacheOpen(source: string, files: any[]) { - const filePath = path.join(getConfig('cache_dir'), source); - await fs.ensureDir(filePath); - if (!files?.length) throw new FormatError('Problem data not found.'); - let etags: Record = {}; - try { - etags = JSON.parse(fs.readFileSync(path.join(filePath, 'etags')).toString()); - } catch (e) { /* ignore */ } - const version = {}; - const filenames = new Set(); - for (const file of files) { - filenames.add(file.name); - version[file.name] = file.etag + file.lastModified; - if (etags[file.name] !== file.etag + file.lastModified) { - await storage.get(`problem/${source}/testdata/${file.name}`, path.join(filePath, file.name)); - } - } - for (const name in etags) { - if (!filenames.has(name) && fs.existsSync(path.join(filePath, name))) await fs.remove(path.join(filePath, name)); - } - fs.writeFileSync(path.join(filePath, 'etags'), JSON.stringify(version)); - fs.writeFileSync(path.join(filePath, 'lastUsage'), new Date().getTime().toString()); - await processTestdata(filePath); - return filePath; - } - - function getNext(that) { - return (data: Partial) => { - logger.debug('Next: %d %o', data); - data.rid = new ObjectID(that.rid); - if (data.case) data.case.message ||= ''; - _judge.next(data); - }; - } - - function getEnd(rid: string) { - return (data: Partial) => { - data.key = 'end'; - data.rid = new ObjectID(rid); - logger.info('End: status=%d score=%d time=%dms memory=%dkb', data.status, data.score, data.time, data.memory); - _judge.end(data); - }; - } - - function getLang(lang: string, doThrow = true) { - if (setting.langs[lang]) return setting.langs[lang]; - if (lang === 'cpp' && setting.langs['cc']) return setting.langs['cc']; - if (doThrow) throw new SystemError('Unsupported language {0}.', [lang]); - return null; - } - - class JudgeTask { - stat: any; - request: any; - event: string; - source: string; - rid: string; - lang: string; - code: CopyInFile; - data: any[]; - config: any; - env: any; - input?: string; - next: ReturnType; - end: ReturnType; - tmpdir: string; - clean: (() => Promise)[]; - folder: string; - getLang: (lang: string) => any; - - constructor(request) { - this.stat = {}; - this.stat.receive = new Date(); - this.request = request; - this.getLang = getLang; - logger.debug('%o', request); - } - - async handle() { - try { - this.stat.handle = new Date(); - this.rid = this.request.rid.toString(); - this.lang = this.request.lang; - this.code = { content: this.request.code }; - this.data = this.request.data; - this.source = this.request.source; - this.config = this.request.config; - this.input = this.request.input; - let tid = this.request.contest?.toString() || ''; - if (tid === '000000000000000000000000') tid = ''; - this.env = { - HYDRO_DOMAIN: this.request.domainId.toString(), - HYDRO_RECORD: this.rid, - HYDRO_LANG: this.lang, - HYDRO_USER: this.request.uid.toString(), - HYDRO_CONTEST: tid, - }; - this.next = getNext(this); - this.end = getEnd(this.rid); - this.tmpdir = path.resolve(getConfig('tmp_dir'), this.rid); - this.clean = []; - fs.ensureDirSync(this.tmpdir); - tmpfs.mount(this.tmpdir, getConfig('tmpfs_size')); - logger.info(`Submission: ${this.source}/${this.rid}`); - await this.submission(); - } catch (e) { - if (e instanceof CompileError) { - this.next({ compilerText: compilerText(e.stdout, e.stderr) }); - this.end({ - status: STATUS.STATUS_COMPILE_ERROR, score: 0, time: 0, memory: 0, - }); - } else if (e instanceof FormatError) { - this.next({ message: { message: e.message, params: e.params } }); - this.end({ - status: STATUS.STATUS_FORMAT_ERROR, score: 0, time: 0, memory: 0, - }); - } else { - logger.error(e); - this.next({ message: { message: e.message, params: e.params, ...process.env.DEV ? { stack: e.stack } : {} } }); - this.end({ - status: STATUS.STATUS_SYSTEM_ERROR, score: 0, time: 0, memory: 0, - }); - } - } - // eslint-disable-next-line no-await-in-loop - for (const clean of this.clean) await clean().catch(noop); - tmpfs.umount(this.tmpdir); - fs.removeSync(this.tmpdir); - } - - async submission() { - this.stat.cache_start = new Date(); - this.folder = await cacheOpen(this.source, this.data); - if ((this.code as any).content.startsWith('@@hydro_submission_file@@')) { - const id = (this.code as any).content.split('@@hydro_submission_file@@')[1]?.split('#')?.[0]; - if (!id) throw new SystemError('Submission File Not Found'); - const target = await fetchCodeFile(id); - this.code = { src: target }; - this.clean.push(() => fs.remove(target)); - } - this.stat.read_cases = new Date(); - this.config = await readCases( - this.folder, - { detail: system.get('hydrojudge.detail'), ...this.config }, - { next: this.next, key: md5(`${this.source}/${getConfig('secret')}`) }, - ); - this.stat.judge = new Date(); - const type = typeof this.input === 'string' ? 'run' : this.config.type || 'default'; - if (!judge[type]) throw new FormatError('Unrecognized problemType: {0}', [type]); - await judge[type].judge(this); - } - } - - task.consume({ type: 'judge' }, (t) => (new JudgeTask(t)).handle().catch(logger.error)); - task.consume({ type: 'judge', priority: { $gt: -50 } }, (t) => (new JudgeTask(t)).handle().catch(logger.error)); +if (process.env.NODE_APP_INSTANCE === '0') { + bus.once('app/started', () => require('./hosts/builtin').postInit()); } - -global.Hydro.service.bus.once('app/started', postInit); diff --git a/packages/hydrojudge/src/sysinfo.ts b/packages/hydrojudge/src/sysinfo.ts deleted file mode 100644 index f76a16d98..000000000 --- a/packages/hydrojudge/src/sysinfo.ts +++ /dev/null @@ -1,80 +0,0 @@ -// TODO this should be moved into web -import path from 'path'; -import fs from 'fs-extra'; -import { noop } from 'lodash'; -import { get as _get } from '@hydrooj/utils/lib/sysinfo'; -import { getConfig } from './config'; -import { Context } from './judge/interface'; -import { judge } from './judge/run'; -import * as tmpfs from './tmpfs'; - -export { update } from '@hydrooj/utils/lib/sysinfo'; - -async function stackSize() { - let output = ''; - try { - const context: Context = { - lang: 'cc', - request: {}, - code: { - content: `#include -using namespace std; -int i=1; -int main(){ - char a[1048576]={'1'}; - cout<<" "<256) return 0; - main(); -}`, - }, - config: { - type: 'default', - time: '3s', - memory: '256m', - user_extra_files: [], - judge_extra_files: [], - }, - stat: {}, - clean: [], - next: (data) => { - if (data.case) output = data.case.message; - }, - end: () => { }, - getLang: () => ({ - compile: '/usr/bin/g++ -Wall -std=c++14 -o foo foo.cc -lm', - execute: '/w/foo', - code_file: 'foo.cc', - highlight: 'cpp', - monaco: 'cpp', - display: 'C++', - time_limit_rate: 1, - domain: [], - key: '', - hidden: false, - }), - tmpdir: path.resolve(getConfig('tmp_dir'), 'sysinfo'), - }; - fs.ensureDirSync(context.tmpdir); - tmpfs.mount(context.tmpdir, '32m'); - await judge(context).catch((e) => console.error(e)); - // eslint-disable-next-line no-await-in-loop - for (const clean of context.clean) await clean().catch(noop); - tmpfs.umount(context.tmpdir); - fs.removeSync(context.tmpdir); - } catch (e) { - return -1; - } - const a = output.split(' '); - return parseInt(a[a.length - 2], 10); -} - -let stackPromise; - -export async function get() { - const info = await _get(); - if (!stackPromise) stackPromise = stackSize(); - // @ts-ignore - info.stack = await stackPromise; - return info; -} diff --git a/packages/hydrojudge/src/task.ts b/packages/hydrojudge/src/task.ts new file mode 100644 index 000000000..cc6128e95 --- /dev/null +++ b/packages/hydrojudge/src/task.ts @@ -0,0 +1,132 @@ +import path from 'path'; +import fs from 'fs-extra'; +import { noop } from 'lodash'; +import { LangConfig } from '@hydrooj/utils/lib/lang'; +import { STATUS } from '@hydrooj/utils/lib/status'; +import type { JudgeResultBody } from 'hydrooj'; +import { Logger } from 'hydrooj/src/logger'; +import readCases from './cases'; +import { getConfig } from './config'; +import { CompileError, FormatError, SystemError } from './error'; +import judge from './judge'; +import { CopyInFile } from './sandbox'; +import * as tmpfs from './tmpfs'; +import { compilerText, Lock, md5 } from './utils'; + +type NextFunction = (data: Partial) => void; +interface Session { + getLang: (name: string) => LangConfig; + getNext: (task: JudgeTask) => NextFunction; + getEnd: (task: JudgeTask) => NextFunction; + cacheOpen: (source: string, files: any[], next?: NextFunction) => Promise; + fetchCodeFile: (target: string) => Promise; + config: { detail: boolean, host?: string }; +} + +const logger = new Logger('judge'); + +export class JudgeTask { + stat: Record; + source: string; + rid: string; + lang: string; + code: CopyInFile; + tmpdir: string; + input?: string; + clean: (() => Promise)[]; + data: any[]; + folder: string; + config: any; + next: (data: Partial) => void; + end: (data: Partial) => void; + env: Record; + + constructor(public session: Session, public request) { + this.stat = {}; + this.stat.receive = new Date(); + logger.debug('%o', request); + } + + async handle() { + const host = this.session.config.host || 'local'; + try { + this.stat.handle = new Date(); + this.rid = this.request.rid.toString(); + this.lang = this.request.lang; + this.code = { content: this.request.code }; + this.data = this.request.data; + this.source = this.request.source; + this.config = this.request.config; + this.input = this.request.input; + let tid = this.request.contest?.toString() || ''; + if (tid === '000000000000000000000000') tid = ''; + this.env = { + HYDRO_DOMAIN: this.request.domainId.toString(), + HYDRO_RECORD: this.rid, + HYDRO_LANG: this.lang, + HYDRO_USER: this.request.uid.toString(), + HYDRO_CONTEST: tid, + }; + this.next = this.session.getNext(this); + this.end = this.session.getEnd(this); + this.tmpdir = path.resolve(getConfig('tmp_dir'), this.rid); + this.clean = []; + await Lock.acquire(`${host}/${this.source}/${this.rid}`); + fs.ensureDirSync(this.tmpdir); + tmpfs.mount(this.tmpdir, getConfig('tmpfs_size')); + logger.info('Submission: %s/%s/%s', host, this.source, this.rid); + await this.doSubmission(); + } catch (e) { + if (e instanceof CompileError) { + this.next({ compilerText: compilerText(e.stdout, e.stderr) }); + this.end({ + status: STATUS.STATUS_COMPILE_ERROR, score: 0, time: 0, memory: 0, + }); + } else if (e instanceof FormatError) { + this.next({ message: 'Testdata configuration incorrect.' }); + this.next({ message: { message: e.message, params: e.params } }); + this.end({ + status: STATUS.STATUS_FORMAT_ERROR, score: 0, time: 0, memory: 0, + }); + } else { + logger.error(e); + this.next({ message: { message: e.message, params: e.params, ...process.env.DEV ? { stack: e.stack } : {} } }); + this.end({ + status: STATUS.STATUS_SYSTEM_ERROR, score: 0, time: 0, memory: 0, + }); + } + } finally { + Lock.release(`${host}/${this.source}/${this.rid}`); + // eslint-disable-next-line no-await-in-loop + for (const clean of this.clean) await clean()?.catch(noop); + tmpfs.umount(this.tmpdir); + fs.removeSync(this.tmpdir); + } + } + + async doSubmission() { + this.stat.cache_start = new Date(); + this.folder = await this.session.cacheOpen(this.source, this.data, this.next); + if ((this.code as any).content.startsWith('@@hydro_submission_file@@')) { + const id = (this.code as any).content.split('@@hydro_submission_file@@')[1]?.split('#')?.[0]; + if (!id) throw new SystemError('Submission File Not Found'); + const target = await this.session.fetchCodeFile(id); + this.code = { src: target }; + this.clean.push(() => fs.remove(target)); + } + this.stat.read_cases = new Date(); + this.config = await readCases( + this.folder, + { + detail: this.session.config.detail, + isSelfSubmission: this.config.problemOwner === this.request.uid, + ...this.config, + }, + { next: this.next, key: md5(`${this.source}/${getConfig('secret')}`) }, + ); + this.stat.judge = new Date(); + const type = typeof this.input === 'string' ? 'run' : this.config.type || 'default'; + if (!judge[type]) throw new FormatError('Unrecognized problemType: {0}', [type]); + await judge[type].judge(this); + } +}