judge: move cacheOpen to task

packages/hydrojudge/src/hosts/builtin.ts

@@ -9,20 +9,28 @@ import {
 } from 'hydrooj';
 import { langs } from 'hydrooj/src/model/setting';
 import { getConfig } from '../config';
-import { FormatError, SystemError } from '../error';
+import { SystemError } from '../error';
 import { compilerVersions, stackSize } from '../info';
+import { Session } from '../interface';
 import { Context } from '../judge/interface';
 import logger from '../log';
 import { versionCheck } from '../sandbox';
 import { JudgeTask } from '../task';
 
-const session = {
+const session: Session = {
     config: { detail: SystemModel.get('hydrojudge.detail') },
-    async fetchFile(name: string) {
-        name = name.split('#')[0];
-        const target = path.join(getConfig('tmp_dir'), name.replace(/\//g, '_'));
-        await StorageModel.get(`submission/${name}`, target);
-        return target;
+    async fetchFile(namespace, files) {
+        if (namespace === null) {
+            const name = Object.keys(files)[0].split('#')[0];
+            const target = path.join(getConfig('tmp_dir'), name.replace(/\//g, '_'));
+            await StorageModel.get(`submission/${name}`, target);
+            return target as any;
+        }
+        for (const key in files) {
+            const target = files[key];
+            await StorageModel.get(`problem/${namespace}/testdata/${key}`, target);
+        }
+        return null;
     },
     getNext(t: Context) {
         t._callbackAwait ||= Promise.resolve();
@@ -51,32 +59,6 @@ const session = {
     async postFile(target: string, filename: string, filepath: string) {
         return await JudgeHandler.processJudgeFileCallback(new ObjectId(target), filename, filepath);
     },
-    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<string, string> = {};
-        try {
-            etags = JSON.parse(await fs.readFile(path.join(filePath, 'etags'), 'utf-8'));
-        } catch (e) { /* ignore */ }
-        const version = {};
-        const filenames = new Set<string>();
-        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 StorageModel.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));
-        }
-        await Promise.all([
-            fs.writeFile(path.join(filePath, 'etags'), JSON.stringify(version)),
-            fs.writeFile(path.join(filePath, 'lastUsage'), Date.now().toString()),
-        ]);
-        return filePath;
-    },
 };
 
 export async function postInit(ctx) {

packages/hydrojudge/src/hosts/hydro.ts

@@ -64,7 +64,6 @@ export default class Hydro implements Session {
     }
 
     async _cacheOpen(source: string, files: any[], next?) {
-        const [domainId, pid] = source.split('/');
         const filePath = path.join(getConfig('cache_dir'), this.config.host, source);
         await fs.ensureDir(filePath);
         if (!files?.length) throw new FormatError('Problem data not found.');
@@ -86,34 +85,38 @@ export default class Hydro implements Session {
         if (filenames.length) {
             log.info(`Getting problem data: ${this.config.host}/${source}`);
             next?.({ message: 'Syncing testdata, please wait...' });
-            await this.ensureLogin();
-            const res = await this.post(`/d/${domainId}/judge/files`, {
-                pid: +pid,
-                files: filenames,
-            });
-            if (!res.body.links) throw new FormatError('problem not exist');
-            const tasks = [];
-            const queue = new PQueue({ concurrency: 10 });
-            for (const name in res.body.links) {
-                tasks.push(queue.add(async () => {
-                    if (name.includes('/')) await fs.ensureDir(path.join(filePath, name.split('/')[0]));
-                    const w = fs.createWriteStream(path.join(filePath, name));
-                    await pipeRequest(this.get(res.body.links[name]), w, 60000, name);
-                }));
-            }
-            await Promise.all(tasks);
+            await this.fetchFile(source, Object.fromEntries(files.map((i) => [i.name, path.join(filePath, i.name)])));
             await fs.writeFile(path.join(filePath, 'etags'), JSON.stringify(version));
         }
         await fs.writeFile(path.join(filePath, 'lastUsage'), new Date().getTime().toString());
         return filePath;
     }
 
-    async fetchFile(name: string) {
-        name = name.split('#')[0];
-        const res = await this.post('judge/files', { id: name });
-        const target = path.join(getConfig('tmp_dir'), name.replace(/\//g, '_'));
-        await pipeRequest(this.get(res.body.url), fs.createWriteStream(target), 60000, name);
-        return target;
+    async fetchFile<T extends string | null>(namespace: T, files: Record<string, string>): Promise<T extends null ? string : null> {
+        if (!namespace) { // record-related resource (code)
+            const name = Object.keys(files)[0].split('#')[0];
+            const res = await this.post('judge/files', { id: name });
+            const target = Object.values(files)[0] || path.join(getConfig('tmp_dir'), Math.random().toString(36).substring(2));
+            await pipeRequest(this.get(res.body.url), fs.createWriteStream(target), 60000, name);
+            return target as any;
+        }
+        const [domainId, pid] = namespace.split('/');
+        await this.ensureLogin();
+        const res = await this.post(`/d/${domainId}/judge/files`, {
+            pid: +pid,
+            files: Object.keys(files),
+        });
+        if (!res.body.links) throw new FormatError('problem not exist');
+        const queue = new PQueue({ concurrency: 10 });
+        for (const name in res.body.links) {
+            queue.add(async () => {
+                if (name.includes('/')) await fs.ensureDir(path.dirname(files[name]));
+                const w = fs.createWriteStream(files[name]);
+                await pipeRequest(this.get(res.body.links[name]), w, 60000, name);
+            });
+        }
+        await queue.onEmpty();
+        return null;
     }
 
     async postFile(target: string, filename: string, file: string, retry = 3) {

packages/hydrojudge/src/interface.ts

@@ -29,8 +29,7 @@ export interface Session {
     getLang: (name: string) => LangConfig;
     getNext: (task: JudgeTask) => NextFunction;
     getEnd: (task: JudgeTask) => NextFunction;
-    cacheOpen: (source: string, files: any[], next?: NextFunction) => Promise<string>;
-    fetchFile: (target: string) => Promise<string>;
+    fetchFile: <T extends null | string>(namespace: T, files: Record<string, string>) => Promise<T extends null ? string : null>;
     postFile: (target: string, filename: string, file: string) => Promise<void>;
     config: { detail: boolean, host?: string, trusted?: boolean };
 }

packages/hydrojudge/src/task.ts

@@ -15,7 +15,7 @@ import {
 import judge from './judge';
 import { Logger } from './log';
 import { CopyIn, CopyInFile, runQueued } from './sandbox';
-import { compilerText, md5 } from './utils';
+import { compilerText, Lock, md5 } from './utils';
 
 const logger = new Logger('judge');
 
@@ -92,11 +92,41 @@ export class JudgeTask {
         }
     }
 
+    async cacheOpen(source: string, files: FileInfo[]) {
+        const lockKey = `${this.session.config?.host || 'local'}/${source}`;
+        await Lock.acquire(lockKey);
+        const filePath = join(getConfig('cache_dir'), lockKey);
+        await fs.ensureDir(filePath);
+        if (!files?.length) throw new FormatError('Problem data not found.');
+        let etags: Record<string, string> = {};
+        try {
+            etags = JSON.parse(await fs.readFile(join(filePath, 'etags'), 'utf-8'));
+        } catch (e) { /* ignore */ }
+        const version = {};
+        const filenames = [];
+        const allFiles = new Set<string>();
+        for (const file of files) {
+            allFiles.add(file.name);
+            version[file.name] = file.etag + file.lastModified;
+            if (etags[file.name] !== file.etag + file.lastModified) filenames.push(file.name);
+        }
+        const allFilesToRemove = Object.keys(etags).filter((name) => !allFiles.has(name) && fs.existsSync(join(filePath, name)));
+        await Promise.all(allFilesToRemove.map((name) => fs.remove(join(filePath, name))));
+        if (filenames.length) {
+            logger.info(`Getting problem data: ${this.session?.config.host || 'local'}/${source}`);
+            this.next({ message: 'Syncing testdata, please wait...' });
+            await this.session.fetchFile(source, Object.fromEntries(files.map((i) => [i.name, join(filePath, i.name)])));
+            await fs.writeFile(join(filePath, 'etags'), JSON.stringify(version));
+        }
+        await fs.writeFile(join(filePath, 'lastUsage'), new Date().getTime().toString());
+        return filePath;
+    }
+
     async doSubmission() {
         this.stat.cache_start = new Date();
-        this.folder = await this.session.cacheOpen(this.source, this.data, this.next);
+        this.folder = await this.cacheOpen(this.source, this.data);
         if (this.files?.code) {
-            const target = await this.session.fetchFile(this.files?.code);
+            const target = await this.session.fetchFile(null, { [this.files.code]: '' });
             this.code = { src: target };
             this.clean.push(() => fs.remove(target));
         }