diff --git a/packages/hydrojudge/package.json b/packages/hydrojudge/package.json index e000488d9..092af368a 100644 --- a/packages/hydrojudge/package.json +++ b/packages/hydrojudge/package.json @@ -1,7 +1,7 @@ { "name": "@hydrooj/hydrojudge", "bin": "bin/hydrojudge.js", - "version": "2.10.11", + "version": "2.10.12", "main": "package.json", "author": "undefined ", "repository": "https://github.com/hydro-dev/Hydro.git", diff --git a/packages/hydrojudge/src/daemon.ts b/packages/hydrojudge/src/daemon.ts index c8b4e95d0..e0df97a31 100644 --- a/packages/hydrojudge/src/daemon.ts +++ b/packages/hydrojudge/src/daemon.ts @@ -48,6 +48,13 @@ process.on('unhandledRejection', (reason, p) => { console.log('Unhandled Rejection at: Promise ', p); }); +async function worker(queue: Queue) { + while ('Orz Soha') { + const [task] = await queue.get(); + task.handle(); + } +} + async function daemon() { const _hosts = getConfig('hosts'); const retry_delay_sec = getConfig('retry_delay_sec'); @@ -59,17 +66,17 @@ async function daemon() { await hosts[i].init(); } global.hosts = hosts; - while ('Orz twd2') { - try { - for (const i in hosts) await hosts[i].consume(queue); - while ('Orz iceb0y') { - const [task] = await queue.get(); - await task.handle(); + worker(queue); + for (const i in hosts) { + while ('Orz twd2') { + try { + await hosts[i].consume(queue); + break; + } catch (e) { + log.error(e, e.stack); + log.info(`在 ${retry_delay_sec} 秒后重试`); + await sleep(retry_delay_sec * 1000); } - } catch (e) { - log.error(e, e.stack); - log.info(`在 ${retry_delay_sec} 秒后重试`); - await sleep(retry_delay_sec * 1000); } } } diff --git a/packages/hydrojudge/src/hosts/hydro.ts b/packages/hydrojudge/src/hosts/hydro.ts index 6a98219d8..5f3def4ab 100644 --- a/packages/hydrojudge/src/hosts/hydro.ts +++ b/packages/hydrojudge/src/hosts/hydro.ts @@ -10,7 +10,9 @@ import { ObjectID } from 'mongodb'; import { LangConfig } from '@hydrooj/utils/lib/lang'; import * as tmpfs from '../tmpfs'; import log from '../log'; -import { compilerText, md5, Queue } from '../utils'; +import { + compilerText, Lock, md5, Queue, +} from '../utils'; import { getConfig } from '../config'; import { FormatError, CompileError, SystemError } from '../error'; import { STATUS_COMPILE_ERROR, STATUS_SYSTEM_ERROR } from '../status'; @@ -69,6 +71,7 @@ class JudgeTask { this.data = this.request.data; this.tmpdir = path.resolve(getConfig('tmp_dir'), this.host, this.rid); this.clean = []; + await Lock.aquire(`${this.host}/${this.domainId}/${this.rid}`); fs.ensureDirSync(this.tmpdir); tmpfs.mount(this.tmpdir, '512m'); log.info('Submission: %s/%s/%s pid=%s', this.host, this.domainId, this.rid, this.pid); @@ -94,10 +97,12 @@ class JudgeTask { status: STATUS_SYSTEM_ERROR, score: 0, time_ms: 0, memory_kb: 0, }); } + } finally { + Lock.release(`${this.host}/${this.domainId}/${this.rid}`); + for (const clean of this.clean) await clean().catch(noop); + tmpfs.umount(this.tmpdir); + fs.removeSync(this.tmpdir); } - for (const clean of this.clean) await clean().catch(noop); - tmpfs.umount(this.tmpdir); - fs.removeSync(this.tmpdir); } async run() { @@ -193,8 +198,18 @@ export default class Hydro { } async cacheOpen(domainId: string, pid: string, files: any[], next?) { + await Lock.aquire(`${this.config.host}/${domainId}/${pid}`); + try { + return this._cacheOpen(domainId, pid, files, next); + } finally { + Lock.release(`${this.config.host}/${domainId}/${pid}`); + } + } + + async _cacheOpen(domainId: string, pid: string, files: any[], next?) { const domainDir = path.join(getConfig('cache_dir'), this.config.host, domainId); - const filePath = path.join(domainDir, pid); await fs.ensureDir(filePath); + const filePath = path.join(domainDir, pid); + await fs.ensureDir(filePath); if (!files?.length) throw new SystemError('Problem data not found.'); let etags: Record = {}; try { diff --git a/packages/hydrojudge/src/utils.ts b/packages/hydrojudge/src/utils.ts index 6f741d14e..ba3834a93 100644 --- a/packages/hydrojudge/src/utils.ts +++ b/packages/hydrojudge/src/utils.ts @@ -4,6 +4,7 @@ import path from 'path'; import { parse } from 'shell-quote'; import _ from 'lodash'; import { EventEmitter } from 'events'; +import { sleep } from '@hydrooj/utils/lib/utils'; import { FormatError } from './error'; const EMPTY_STR = /^[ \r\n\t]*$/i; @@ -35,7 +36,7 @@ export class Queue extends EventEmitter { }); } const items = []; - for (let i = 0; i < count; i++) { items.push(this.queue[i]); } + for (let i = 0; i < count; i++) items.push(this.queue[i]); this.queue = _.drop(this.queue, count); return items as T[]; } @@ -44,7 +45,7 @@ export class Queue extends EventEmitter { this.queue.push(value); if (this.waiting.length && this.waiting[0].count <= this.queue.length) { const items = []; - for (let i = 0; i < this.waiting[0].count; i++) { items.push(this.queue[i]); } + for (let i = 0; i < this.waiting[0].count; i++) items.push(this.queue[i]); this.queue = _.drop(this.queue, this.waiting[0].count); this.waiting[0].resolve(items); this.waiting.shift(); @@ -52,6 +53,19 @@ export class Queue extends EventEmitter { } } +export namespace Lock { + const data = {}; + + export async function aquire(key: string) { + while (data[key]) sleep(100); + data[key] = true; + } + + export function release(key: string) { + data[key] = false; + } +} + export function compilerText(stdout: string, stderr: string) { const ret = []; if (!EMPTY_STR.test(stdout)) ret.push(stdout.substr(0, 1024 * 1024));