judge: add resource lock

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 <i@undefined.moe>",
     "repository": "https://github.com/hydro-dev/Hydro.git",

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<any>) {
+    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);
         }
     }
 }

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<string, string> = {};
         try {

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<T> 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<T> 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<T> 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));