judge: fix #1204

packages/hydrojudge/src/checkers.ts

@@ -16,11 +16,13 @@ export interface CheckConfig {
     env?: Record<string, string>;
 }
 
+export interface PassInfo { input: CopyInFile, state?: Record<string, CopyInFile>, dispose: () => PromiseLike<any> }
+
 type Checker = (config: CheckConfig) => Promise<{
     status: number;
     score: number;
     message: string | { message: string, params?: any[] };
-    nextPass?: { input: CopyInFile, state?: Record<string, CopyInFile> };
+    nextPass?: PassInfo;
 }>;
 
 function parseDiffMsg(msg: string) {
@@ -237,7 +239,7 @@ const checkers: Record<string, Checker> = new Proxy({
     },
 
     async testlib(config) {
-        const { stderr, status, code, fileIds } = await runQueued(`${config.execute} /w/in /w/user_out /w/answer`, {
+        const { stderr, status, code, fileIds, [Symbol.asyncDispose]: dispose } = await runQueued(`${config.execute} /w/in /w/user_out /w/answer`, {
             copyIn: {
                 in: config.input,
                 user_out: config.user_stdout,
@@ -254,6 +256,7 @@ const checkers: Record<string, Checker> = new Proxy({
                 [STATUS.STATUS_TIME_LIMIT_EXCEEDED]: 'Checker Time Limit Exceeded',
                 [STATUS.STATUS_MEMORY_LIMIT_EXCEEDED]: 'Checker Memory Limit Exceeded',
             }[status];
+            dispose();
             return {
                 status: STATUS.STATUS_SYSTEM_ERROR,
                 score: 0,
@@ -261,6 +264,7 @@ const checkers: Record<string, Checker> = new Proxy({
             };
         }
         if (status === STATUS.STATUS_RUNTIME_ERROR && !stderr?.trim()) {
+            dispose();
             return {
                 status: STATUS.STATUS_SYSTEM_ERROR,
                 score: 0,
@@ -274,15 +278,17 @@ const checkers: Record<string, Checker> = new Proxy({
                 nextPass: {
                     input: { fileId: fileIds['nextpass.in'] },
                     state: fileIds['state.txt'] ? { 'state.txt': { fileId: fileIds['state.txt'] } } : undefined,
+                    dispose,
                 },
             };
         }
+        dispose();
         return result;
     },
 
     // https://www.kattis.com/problem-package-format/spec/2023-07-draft.html#output-validator
     async kattis(config) {
-        const { files, fileIds, code } = await runQueued(`${config.execute} input answer_file feedback_dir`, {
+        const { files, fileIds, code, [Symbol.asyncDispose]: dispose } = await runQueued(`${config.execute} input answer_file feedback_dir`, {
             copyIn: {
                 input: config.input,
                 answer_file: config.output,
@@ -328,10 +334,12 @@ const checkers: Record<string, Checker> = new Proxy({
                     state: fileIds['feedback_dir/state.txt']
                         ? { 'feedback_dir/state.txt': { fileId: fileIds['feedback_dir/state.txt'] } }
                         : undefined,
+                    dispose,
                 },
             };
         }
 
+        dispose();
         return { status, score, message };
     },
 }, {

packages/hydrojudge/src/flow.ts

@@ -9,7 +9,7 @@ import { Context, ContextSubTask } from './judge/interface';
 interface Task {
     compile: () => Promise<void>;
     judgeCase: (c: NormalizedCase) => (
-        (ctx: Context, ctxSubtask: ContextSubTask, runner?: Function) => Promise<JudgeResultBody['case']>
+        (ctx: Context, ctxSubtask: ContextSubTask) => Promise<JudgeResultBody['case']>
     );
 }
 
@@ -47,7 +47,7 @@ function judgeSubtask(subtask: NormalizedSubtask, sid: string, judgeCase: Task['
                         message: '',
                     } : await (async () => {
                         using span = ctx.startChildSpan('judge.case', { id: subtask.cases[cid].id, subtaskId: subtask.id });
-                        const r = await runner(ctx, ctxSubtask, runner);
+                        const r = await runner(ctx, ctxSubtask);
                         span.setAttributes({ status: r?.status, time: r?.time, memory: r?.memory });
                         return r;
                     })();
@@ -98,7 +98,7 @@ export const runFlow = async (ctx: Context, task: Task) => {
             score: subtask.type === 'min' ? subtask.score : 0,
         };
         const runner = task.judgeCase(subtask.cases.find((i) => i.input.endsWith(ctx.meta.hackRejudge)));
-        const res = await runner(ctx, ctxSubtask, runner);
+        const res = await runner(ctx, ctxSubtask);
         if (res) ctx.next({ case: res });
         if (res?.status !== STATUS.STATUS_ACCEPTED) {
             const totalScore = Math.sum(ctx.config.subtasks.map((i) => i.score));

packages/hydrojudge/src/judge/default.ts

@@ -1,21 +1,22 @@
+/* eslint-disable no-await-in-loop */
 import { NormalizedCase, STATUS } from '@hydrooj/common';
-import checkers from '../checkers';
+import checkers, { PassInfo } from '../checkers';
 import { runFlow } from '../flow';
 import { runQueued } from '../sandbox';
 import signals from '../signals';
-import { Context, ContextSubTask, MultiPassContext } from './interface';
-
-function judgeCase(c: NormalizedCase) {
-    const mp: MultiPassContext = { i: 0 };
-    return async (ctx: Context, ctxSubtask: ContextSubTask, runner?: Function) => {
-        const { address_space_limit, process_limit } = ctx.session.getLang(ctx.lang);
-        if (ctx.config.multi_pass && !mp.i) mp.i = 1;
+import { Context, ContextSubTask } from './interface';
 
+const judgeCase = (c: NormalizedCase) => async (ctx: Context, ctxSubtask: ContextSubTask) => {
+    const { address_space_limit, process_limit } = ctx.session.getLang(ctx.lang);
+    const maxPasses = ctx.config.multi_pass || 0;
+    let pass = ctx.config.multi_pass ? 1 : 0;
+    let state: PassInfo = { input: { src: c.input }, state: {}, dispose: () => Promise.resolve() };
+    while (true) {
         await using res = await runQueued(
             ctx.execute.execute,
             {
-                stdin: mp.input ?? { src: c.input },
-                copyIn: { ...ctx.execute.copyIn, ...mp.state },
+                stdin: state.input,
+                copyIn: { ...ctx.execute.copyIn, ...state.state },
                 filename: ctx.config.filename,
                 time: c.time,
                 memory: c.memory,
@@ -23,7 +24,7 @@ function judgeCase(c: NormalizedCase) {
                 addressSpaceLimit: address_space_limit,
                 processLimit: process_limit,
             },
-            `judgeCase[${c.id}]${mp.i ? `{pass=${mp.i}}` : ''}<${ctx.rid}>`,
+            `judgeCase[${c.id}${pass ? `#${pass}` : ''}]<${ctx.rid}>`,
         );
         const {
             code, signalled, time, memory, fileIds,
@@ -31,7 +32,7 @@ function judgeCase(c: NormalizedCase) {
         let { status } = res;
         let message: any = '';
         let score = 0;
-        let nextPass: any;
+        let nextPass;
         if (status === STATUS.STATUS_ACCEPTED) {
             if (time > c.time) {
                 status = STATUS.STATUS_TIME_LIMIT_EXCEEDED;
@@ -44,10 +45,10 @@ function judgeCase(c: NormalizedCase) {
                     execute: ctx.checker.execute,
                     copyIn: {
                         ...ctx.checker.copyIn,
-                        ...mp.state ?? {},
+                        ...state.state,
                     },
                     code: ctx.code,
-                    input: mp.input ?? { src: c.input },
+                    input: state.input,
                     output: { src: c.output },
                     user_stdout: fileIds.stdout ? { fileId: fileIds.stdout } : { content: '' },
                     user_stderr: fileIds.stderr ? { fileId: fileIds.stderr } : { content: '' },
@@ -58,35 +59,36 @@ function judgeCase(c: NormalizedCase) {
                         HYDRO_TESTCASE: c.id.toString(),
                         HYDRO_TIME_USAGE: time.toString(),
                         HYDRO_MEMORY_USAGE: Math.floor(memory / 1024).toString(),
-                        ...(mp.i ? { HYDRO_MULTI_PASS: mp.i.toString() } : {}),
+                        ...(pass ? { HYDRO_MULTI_PASS: pass.toString() } : {}),
                     },
                 }));
-                if (mp.i && typeof message === 'string') message = `${message} [Pass ${mp.i}]`;
             }
         } else if (status === STATUS.STATUS_RUNTIME_ERROR && code && ctx.config.detail === 'full') {
             if (code < 32 && signalled) message = signals[code];
-            else message = { message: 'Your program returned {0}.', params: [`${code}${mp.i ? ` [Pass ${mp.i}]` : ''}`] };
-            if (mp.i && typeof message === 'string') message = `${message} [Pass ${mp.i}]`;
+            else message = { message: 'Your program returned {0}.', params: [`${code}${pass ? ` [Pass ${pass}]` : ''}`] };
         }
+        if (pass && typeof message === 'string' && message) message += ` [Pass ${pass}]`;
         if (nextPass) {
-            if (mp.i < ctx.config.multi_pass) {
-                mp.input = nextPass.input;
-                mp.state = nextPass.state ?? undefined;
-                mp.i++;
-                return await runner(ctx, ctxSubtask, runner);
+            if (pass < maxPasses) {
+                pass++;
+                state.dispose();
+                state = nextPass;
+                continue;
             }
+            nextPass.dispose();
             status = STATUS.STATUS_SYSTEM_ERROR;
             score = 0;
             message = { message: 'Exceeded maximum number of passes ({0}).', params: [ctx.config.multi_pass] };
         }
-        if (runner && ctx.rerun && c.time <= 5000 && status === STATUS.STATUS_TIME_LIMIT_EXCEEDED) {
+        if (ctx.rerun && c.time <= 5000 && status === STATUS.STATUS_TIME_LIMIT_EXCEEDED) {
             ctx.rerun--;
-            return await runner(ctx, ctxSubtask, runner);
+            continue;
         }
-        if (!ctx.request.rejudged && !ctx.analysis && [STATUS.STATUS_WRONG_ANSWER, STATUS.STATUS_RUNTIME_ERROR].includes(status)) {
+        if (!ctx.request.rejudged && !ctx.analysis && !pass && [STATUS.STATUS_WRONG_ANSWER, STATUS.STATUS_RUNTIME_ERROR].includes(status)) {
             ctx.analysis = true;
-            await ctx.runAnalysis(ctx.execute, { src: c.input });
+            await ctx.runAnalysis(ctx.execute, state.input);
         }
+        state.dispose();
         return {
             id: c.id,
             subtaskId: ctxSubtask.subtask.id,
@@ -96,8 +98,8 @@ function judgeCase(c: NormalizedCase) {
             memory,
             message,
         };
-    };
-}
+    }
+};
 
 export const judge = async (ctx: Context) => await runFlow(ctx, {
     compile: async () => {

packages/hydrojudge/src/judge/interactive.ts

@@ -1,22 +1,24 @@
+/* eslint-disable no-await-in-loop */
 import { NormalizedCase, STATUS } from '@hydrooj/common';
+import { PassInfo } from '../checkers';
 import { runFlow } from '../flow';
 import { runPiped } from '../sandbox';
 import signals from '../signals';
 import { parse } from '../testlib';
-import { Context, ContextSubTask, MultiPassContext } from './interface';
-
-function judgeCase(c: NormalizedCase) {
-    const mp: MultiPassContext = { i: 0 };
-    return async (ctx: Context, ctxSubtask: ContextSubTask, runner?: Function) => {
-        const { address_space_limit, process_limit } = ctx.session.getLang(ctx.lang);
-        if (ctx.config.multi_pass && !mp.i) mp.i = 1;
+import { Context, ContextSubTask } from './interface';
 
+const judgeCase = (c: NormalizedCase) => async (ctx: Context, ctxSubtask: ContextSubTask) => {
+    const { address_space_limit, process_limit } = ctx.session.getLang(ctx.lang);
+    const maxPasses = ctx.config.multi_pass || 0;
+    let pass = ctx.config.multi_pass ? 1 : 0;
+    let state: PassInfo = { input: { src: c.input }, state: {}, dispose: () => Promise.resolve() };
+    while (true) {
         const [{
             code, signalled, time, memory,
         }, resInteractor] = await runPiped([
             {
                 execute: ctx.executeUser.execute,
-                copyIn: { ...ctx.executeUser.copyIn, ...mp.state },
+                copyIn: { ...ctx.executeUser.copyIn, ...state.state },
                 time: c.time,
                 memory: c.memory,
                 addressSpaceLimit: address_space_limit,
@@ -25,10 +27,10 @@ function judgeCase(c: NormalizedCase) {
             {
                 execute: `${ctx.executeInteractor.execute} /w/in /w/tout /w/out`,
                 copyIn: {
-                    in: mp.input ?? { src: c.input },
+                    in: state.input,
                     out: c.output ? { src: c.output } : { content: '' },
                     ...ctx.executeInteractor.copyIn,
-                    ...mp.state ?? {},
+                    ...state.state,
                 },
                 time: c.time * 2,
                 memory: c.memory * 2,
@@ -37,17 +39,18 @@ function judgeCase(c: NormalizedCase) {
                 env: {
                     ...ctx.env,
                     HYDRO_TESTCASE: c.id.toString(),
-                    ...(mp.i ? { HYDRO_MULTI_PASS: mp.i.toString() } : {}),
+                    ...(pass ? { HYDRO_MULTI_PASS: pass.toString() } : {}),
                 },
             },
         ], [
             { in: { index: 0, fd: 1 }, out: { index: 1, fd: 0 }, name: 'userToInteractor' },
             { in: { index: 1, fd: 1 }, out: { index: 0, fd: 0 }, name: 'interactorToUser' },
-        ], undefined, `judgeCase[${c.id}]${mp.i ? `[pass=${mp.i}]` : ''}<${ctx.rid}>`);
+        ], undefined, `judgeCase[${c.id}${pass ? `#${pass}` : ''}]<${ctx.rid}>`);
+        const dispose = resInteractor[Symbol.asyncDispose];
         // TODO handle tout (maybe pass to checker?)
         let status: number;
         let score = 0;
-        let message: any = mp.i ? `[Pass ${mp.i}] ` : '';
+        let message: any = pass ? `[Pass ${pass}] ` : '';
         if (time > c.time) {
             status = STATUS.STATUS_TIME_LIMIT_EXCEEDED;
         } else if (memory > c.memory * 1024) {
@@ -61,21 +64,27 @@ function judgeCase(c: NormalizedCase) {
             status = result.status;
             score = result.score;
             message += result.message;
-            if (resInteractor.code && !(resInteractor.stderr || '').trim().length) message += ` (Interactor exited with code ${resInteractor.code})`;
-            if (status === STATUS.STATUS_ACCEPTED) {
-                if (resInteractor.fileIds['nextpass.in']) {
-                    if (mp.i < ctx.config.multi_pass) {
-                        mp.input = { fileId: resInteractor.fileIds['nextpass.in'] };
-                        mp.state = resInteractor.fileIds['state.txt'] ? { 'state.txt': { fileId: resInteractor.fileIds['state.txt'] } } : undefined;
-                        mp.i++;
-                        return await runner!(ctx, ctxSubtask, runner);
-                    }
-                    status = STATUS.STATUS_SYSTEM_ERROR;
-                    score = 0;
-                    message = { message: 'Exceeded maximum number of passes ({0}).', params: [ctx.config.multi_pass] };
+            if (resInteractor.code && !(resInteractor.stderr || '').trim().length) {
+                message += ` (Interactor exited with code ${resInteractor.code})`;
+            }
+            if (status === STATUS.STATUS_ACCEPTED && resInteractor.fileIds['nextpass.in']) {
+                if (pass < maxPasses) {
+                    state.dispose();
+                    state = {
+                        input: { fileId: resInteractor.fileIds['nextpass.in'] },
+                        state: resInteractor.fileIds['state.txt'] ? { 'state.txt': { fileId: resInteractor.fileIds['state.txt'] } } : undefined,
+                        dispose,
+                    };
+                    pass++;
+                    continue;
                 }
+                status = STATUS.STATUS_SYSTEM_ERROR;
+                score = 0;
+                message = { message: 'Exceeded maximum number of passes ({0}).', params: [ctx.config.multi_pass] };
             }
         }
+        dispose();
+        state.dispose();
         return {
             id: c.id,
             subtaskId: ctxSubtask.subtask.id,
@@ -85,8 +94,8 @@ function judgeCase(c: NormalizedCase) {
             memory,
             message,
         };
-    };
-}
+    }
+};
 
 export const judge = async (ctx: Context) => await runFlow(ctx, {
     compile: async () => {

packages/hydrojudge/src/judge/interface.ts

@@ -1,7 +1,6 @@
 import PQueue from 'p-queue';
 import { NormalizedSubtask } from '@hydrooj/common';
 import { Execute } from '../interface';
-import { CopyInFile } from '../sandbox';
 import { JudgeTask } from '../task';
 
 export type Context = JudgeTask & RuntimeContext;
@@ -30,9 +29,3 @@ export interface ContextSubTask {
     score: number;
     status: number;
 }
-
-export interface MultiPassContext {
-    i: number;
-    input?: CopyInFile;
-    state?: Record<string, CopyInFile>;
-}

packages/hydrojudge/src/sandbox.ts

@@ -47,7 +47,7 @@ export interface Parameter {
     filename?: string;
 }
 
-interface SandboxAdaptedResult {
+interface SandboxAdaptedResult extends AsyncDisposable {
     status: number;
     code: number;
     signalled: boolean;
@@ -74,6 +74,10 @@ function parseArgs(execute: string): string[] {
     return args;
 }
 
+export async function del(fileId: string) {
+    await client.deleteFile(fileId);
+}
+
 function proc(params: Parameter): Cmd {
     const copyOut = supportOptional
         ? (params.copyOut || [])
@@ -126,6 +130,7 @@ function adaptResult(result: SandboxResult, params: Parameter): SandboxAdaptedRe
         memory: result.memory / 1024,
         files: result.files,
         code: result.exitStatus,
+        [Symbol.asyncDispose]: () => Promise.allSettled([...new Set(Object.values(result.fileIds || {}))].map(del)) as Promise<any>,
     };
     if (ret.time > (params.time || 16000)) {
         ret.status = STATUS.STATUS_TIME_LIMIT_EXCEEDED;
@@ -182,10 +187,6 @@ export async function runPiped(
     return res.map((r) => adaptResult(r, params)) as SandboxAdaptedResult[];
 }
 
-export async function del(fileId: string) {
-    await client.deleteFile(fileId);
-}
-
 export async function get(fileId: string, dest?: string) {
     return await client.getFile(fileId, dest);
 }
@@ -209,8 +210,9 @@ export function runQueued(
         : [arg0, arg1, arg2 || {}, arg3 || '', arg4 || 0];
     return queue.add(async () => {
         const res = await runPiped(execute, pipeMapping, params, trace);
+        const disposers = res.map((t) => t[Symbol.asyncDispose]);
         const ret = single ? res[0] : res;
-        (ret as any)[Symbol.asyncDispose] = () => Promise.allSettled(res.flatMap((t) => Object.values(t.fileIds || {}).map(del)));
+        (ret as any)[Symbol.asyncDispose] = () => Promise.allSettled(disposers.map((dispose) => dispose()));
         return ret;
     }, { priority });
 }