diff --git a/evalscope/service/blueprints/eval.py b/evalscope/service/blueprints/eval.py index 8235d176f..c1e0eda67 100644 --- a/evalscope/service/blueprints/eval.py +++ b/evalscope/service/blueprints/eval.py @@ -13,6 +13,7 @@ DEFAULT_MULTIMODAL_BENCHMARKS, DEFAULT_TEXT_BENCHMARKS, OUTPUT_DIR, + TaskStoppedError, build_benchmark_entry, create_log_file, discover_all_benchmarks, @@ -163,6 +164,9 @@ def _execute_task(task_id: str, task_config: TaskConfig, label: str = 'Task'): 'result': serialize_result(result), 'table': table_str }) + except TaskStoppedError: + logger.info(f'[{task_id}] {label} stopped by user.') + return jsonify({'status': 'stopped', 'task_id': task_id}) except Exception as e: logger.error(f'[{task_id}] {label} failed: {e}') return jsonify({'status': 'error', 'task_id': task_id, 'error': str(e)}), 500 diff --git a/evalscope/service/blueprints/perf.py b/evalscope/service/blueprints/perf.py index a2d307120..610e7e649 100644 --- a/evalscope/service/blueprints/perf.py +++ b/evalscope/service/blueprints/perf.py @@ -11,6 +11,7 @@ from ..perf_archive import PerfArchiveError from ..utils import ( OUTPUT_DIR, + TaskStoppedError, create_log_file, get_log_content, run_in_subprocess, @@ -104,6 +105,9 @@ def run_performance_test(): 'result': serialize_result(result), 'table': table_str }) + except TaskStoppedError: + logger.info(f'[{task_id}] Task stopped by user.') + return jsonify({'status': 'stopped', 'task_id': task_id}) except Exception as e: logger.error(f'[{task_id}] Task failed: {e}') return jsonify({'status': 'error', 'task_id': task_id, 'error': str(e)}), 500 diff --git a/evalscope/service/utils/__init__.py b/evalscope/service/utils/__init__.py index ff4401510..80bee3a69 100644 --- a/evalscope/service/utils/__init__.py +++ b/evalscope/service/utils/__init__.py @@ -9,7 +9,14 @@ parse_benchmark_description, ) from .log import OUTPUT_DIR, create_log_file, get_log_content, validate_task_id -from .process import run_eval_wrapper, run_in_subprocess, run_perf_wrapper, serialize_result, stop_process +from .process import ( + TaskStoppedError, + run_eval_wrapper, + run_in_subprocess, + run_perf_wrapper, + serialize_result, + stop_process, +) __all__ = [ 'OUTPUT_DIR', @@ -21,6 +28,7 @@ 'serialize_result', 'stop_process', 'run_in_subprocess', + 'TaskStoppedError', 'DEFAULT_TEXT_BENCHMARKS', 'DEFAULT_MULTIMODAL_BENCHMARKS', 'build_benchmark_entry', diff --git a/evalscope/service/utils/process.py b/evalscope/service/utils/process.py index 5175dd85e..fbbdf0b1f 100644 --- a/evalscope/service/utils/process.py +++ b/evalscope/service/utils/process.py @@ -23,6 +23,26 @@ _active_lock = threading.Lock() +# --------------------------------------------------------------------------- +# Track tasks stopped by the user so we can distinguish intentional stop +# from a genuine subprocess crash. +# --------------------------------------------------------------------------- + +_user_stopped_tasks: set[str] = set() +"""Task IDs that were explicitly stopped by the user via /stop.""" + + +class TaskStoppedError(Exception): + """Raised when a task is explicitly stopped by the user. + + This is a *normal* termination, not a crash — callers should handle it + gracefully (e.g. return ``{'status': 'stopped'}`` instead of 500). + """ + + def __init__(self, task_id: str): + self.task_id = task_id + super().__init__(f'Task {task_id} was stopped by the user.') + def register_process(task_id: str, proc: multiprocessing.Process) -> None: """Register a running subprocess so it can be stopped later.""" @@ -45,12 +65,21 @@ def stop_process(task_id: str) -> bool: proc = _active_processes.pop(task_id, None) if proc is None: return False + if not proc.is_alive(): + # Process already exited on its own; do NOT mark as user-stopped, or a + # pending success result would be masked as TaskStoppedError. Treat the + # stop request as a no-op. + logger.info(f'Task {task_id} already finished before stop; ignoring stop request.') + return False + # Mark as user-stopped BEFORE terminating: the parent thread blocked in + # run_in_subprocess polls the child every 0.1s and would otherwise observe + # the death first and misreport it as a crash. + _user_stopped_tasks.add(task_id) + proc.terminate() + proc.join(timeout=3) if proc.is_alive(): - proc.terminate() - proc.join(timeout=3) - if proc.is_alive(): - proc.kill() - proc.join(timeout=2) + proc.kill() + proc.join(timeout=2) logger.info(f'Task {task_id} stopped by user.') return True @@ -137,27 +166,40 @@ def run_in_subprocess(func, *args, task_id=None, **kwargs): if task_id: unregister_process(task_id) + # Consume the user-stop marker exactly once so it never leaks, regardless + # of which branch below handles the outcome. + user_stopped = bool(task_id) and task_id in _user_stopped_tasks + if task_id: + _user_stopped_tasks.discard(task_id) + + if res is None: + # The child exited without the poll loop seeing a result. Do one final + # non-blocking drain in case the item arrived between the last loop + # iteration and p.join() returning — a produced result always takes + # priority over the user-stop marker. + try: + res = result_queue.get_nowait() + except queue.Empty: + res = None + if res is not None: if res['status'] == 'error': + # A user-initiated stop may surface as an error result when the + # child converts SIGTERM into an exception — still a graceful stop. + if user_stopped: + raise TaskStoppedError(task_id) stderr_info = res.get('stderr', '') stderr_section = f'\n[stderr]\n{stderr_info}' if stderr_info.strip() else '' raise RuntimeError(f"Subprocess error: {res['error']}\n{res.get('traceback', '')}{stderr_section}") return res['result'] + # If the task was explicitly stopped by the user, treat it as a graceful + # termination — not a crash. + if user_stopped: + raise TaskStoppedError(task_id) + # res is still None: the child exited without putting anything in the queue # (OOM, SIGKILL, import error, segfault, etc.). - # Do one final non-blocking check in case the item arrived between the last - # loop iteration and p.join() returning. - try: - res = result_queue.get_nowait() - if res['status'] == 'error': - stderr_info = res.get('stderr', '') - stderr_section = f'\n[stderr]\n{stderr_info}' if stderr_info.strip() else '' - raise RuntimeError(f"Subprocess error: {res['error']}\n{res.get('traceback', '')}{stderr_section}") - return res['result'] - except queue.Empty: - pass - raise RuntimeError( f'Subprocess terminated unexpectedly (exit code {p.exitcode}). ' 'The child process may have crashed due to OOM, a missing import, ' diff --git a/evalscope/web/src/api/schemas/eval.schema.ts b/evalscope/web/src/api/schemas/eval.schema.ts index d7c876f8c..722a44f7d 100644 --- a/evalscope/web/src/api/schemas/eval.schema.ts +++ b/evalscope/web/src/api/schemas/eval.schema.ts @@ -29,6 +29,8 @@ export const progressResponseSchema = z .object({ percent: z.number(), current_step: z.string().optional(), + /** Tracker lifecycle state written by ProgressTracker: running | completed | error. */ + status: z.string().optional(), }) .catchall(z.unknown()) diff --git a/evalscope/web/src/components/perf/PerfConfigForm.tsx b/evalscope/web/src/components/perf/PerfConfigForm.tsx index 9f0b42165..4f4a1c518 100644 --- a/evalscope/web/src/components/perf/PerfConfigForm.tsx +++ b/evalscope/web/src/components/perf/PerfConfigForm.tsx @@ -1,9 +1,11 @@ import { useState, type SyntheticEvent } from 'react' import { useLocale } from '@/contexts/LocaleContext' import Button from '@/components/ui/Button' +import Card from '@/components/ui/Card' import Field from '@/components/ui/Field' import { FORM_INPUT_CLASS, inputClass } from '@/components/ui/formStyles' import { computeFirstInvalid, validateNumeric, FORM_MESSAGE_KEYS } from '@/domain/form/validation' +import { ChevronDown, ChevronUp } from 'lucide-react' import { useFormErrors } from '@/hooks/useFormErrors' import { ApiKeyField, ApiUrlField, ModelField } from '@/components/tasks/TaskFormFields' @@ -26,6 +28,7 @@ const IDS = { dataset: 'perf-dataset', maxPromptLen: 'perf-maxPromptLen', minPromptLen: 'perf-minPromptLen', + tokenizerPath: 'perf-tokenizerPath', } as const /** DOM order of focusable fields, drives first-invalid focus on submit. */ @@ -42,6 +45,7 @@ const DOM_ORDER: string[] = [ IDS.dataset, IDS.maxPromptLen, IDS.minPromptLen, + IDS.tokenizerPath, ] export default function PerfConfigForm({ onSubmit, disabled }: Props) { @@ -58,6 +62,8 @@ export default function PerfConfigForm({ onSubmit, disabled }: Props) { const [dataset, setDataset] = useState('') const [maxPromptLen, setMaxPromptLen] = useState('') const [minPromptLen, setMinPromptLen] = useState('') + const [tokenizerPath, setTokenizerPath] = useState('') + const [showMore, setShowMore] = useState(false) const { setErrors, errorFor: errMsg, clearError: clearErr } = useFormErrors() @@ -119,6 +125,7 @@ export default function PerfConfigForm({ onSubmit, disabled }: Props) { if (dataset) config.dataset = dataset if (maxPromptLen) config.max_prompt_length = Number(maxPromptLen) if (minPromptLen) config.min_prompt_length = Number(minPromptLen) + if (tokenizerPath) config.tokenizer_path = tokenizerPath onSubmit(config) } @@ -194,6 +201,28 @@ export default function PerfConfigForm({ onSubmit, disabled }: Props) { + {/* ── More params toggle ── */} + + + {showMore && ( + +
+ + {(aria) => ( + setTokenizerPath(e.target.value)} className={FORM_INPUT_CLASS} placeholder="/path/to/tokenizer" /> + )} + +
+
+ )} + diff --git a/evalscope/web/src/components/tasks/TaskRunnerPage.tsx b/evalscope/web/src/components/tasks/TaskRunnerPage.tsx index f4f1619cc..a389d73bd 100644 --- a/evalscope/web/src/components/tasks/TaskRunnerPage.tsx +++ b/evalscope/web/src/components/tasks/TaskRunnerPage.tsx @@ -91,7 +91,11 @@ export default function TaskRunnerPage({ interval: 5000, onData: (data) => { setProgress(data.percent ?? 0) - if (data.percent >= 100) setRunning(false) + // Stop polling when benchmark finishes (via percent) or when the + // backend progress tracker reports error / completed (status key). + if (data.percent >= 100 || data.status === 'error' || data.status === 'completed') { + setRunning(false) + } }, }) diff --git a/evalscope/web/src/i18n/translations/perf.ts b/evalscope/web/src/i18n/translations/perf.ts index 528a6f50d..531d82215 100644 --- a/evalscope/web/src/i18n/translations/perf.ts +++ b/evalscope/web/src/i18n/translations/perf.ts @@ -12,6 +12,8 @@ export const en: Dict = { dataset: 'Test Dataset', maxPromptLen: 'Max Prompt Length', minPromptLen: 'Min Prompt Length', + moreParams: 'More Parameters', + tokenizerPath: 'Tokenizer Path', startPerf: 'Start Performance Test', status: 'Status & Logs', ready: 'Status: Ready', @@ -29,6 +31,8 @@ export const zh: Dict = { dataset: '测试数据集', maxPromptLen: '最大Prompt长度', minPromptLen: '最小Prompt长度', + moreParams: '更多参数', + tokenizerPath: 'Tokenizer 路径', startPerf: '开始性能测试', status: '运行状态与日志', ready: '当前状态: 准备就绪', diff --git a/tests/service/__init__.py b/tests/service/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/service/test_process_stop.py b/tests/service/test_process_stop.py new file mode 100644 index 000000000..2a58d1dfd --- /dev/null +++ b/tests/service/test_process_stop.py @@ -0,0 +1,84 @@ +# Copyright (c) Alibaba, Inc. and its affiliates. +"""Tests for user-initiated stop handling in service subprocess utilities. + +Covers the interaction between ``run_in_subprocess`` and ``stop_process``: +a user-initiated stop must surface as :class:`TaskStoppedError` (graceful), +while a genuine child crash must still raise ``RuntimeError``. +""" +import multiprocessing +import threading +import time +import unittest + +from evalscope.service.utils.process import ( + TaskStoppedError, + _user_stopped_tasks, + register_process, + run_in_subprocess, + stop_process, +) + + +def _sleep_forever() -> str: + time.sleep(60) + return 'done' + + +def _quick_success() -> str: + return 'ok' + + +def _crash() -> None: + raise ValueError('boom') + + +class TestProcessStop(unittest.TestCase): + + def test_user_stop_raises_task_stopped_error(self): + """Stopping a running task raises TaskStoppedError, not RuntimeError.""" + task_id = 'test-stop-graceful' + outcome = {} + + def run() -> None: + try: + outcome['result'] = run_in_subprocess(_sleep_forever, task_id=task_id) + except Exception as e: # noqa: BLE001 - capture for assertion + outcome['exc'] = e + + t = threading.Thread(target=run) + t.start() + time.sleep(3) # let the spawned child start sleeping + + self.assertTrue(stop_process(task_id)) + t.join(timeout=30) + + self.assertIsInstance(outcome.get('exc'), TaskStoppedError) + + def test_success_returns_result(self): + """A normal run returns the child's result unchanged.""" + self.assertEqual(run_in_subprocess(_quick_success, task_id='test-stop-success'), 'ok') + + def test_child_crash_still_raises_runtime_error(self): + """A genuine child error is not misreported as a user stop.""" + with self.assertRaises(RuntimeError): + run_in_subprocess(_crash, task_id='test-stop-crash') + + def test_stop_unknown_task_returns_false(self): + """Stopping a non-existent task is a no-op returning False.""" + self.assertFalse(stop_process('test-stop-nonexistent')) + + def test_stop_after_natural_exit_is_noop(self): + """Stopping an already-exited (but still registered) process must not + set the user-stop marker, so a pending success result is not masked.""" + task_id = 'test-stop-already-dead' + proc = multiprocessing.get_context('spawn').Process(target=_quick_success) + proc.start() + proc.join() # let it exit naturally before the stop request + register_process(task_id, proc) + + self.assertFalse(stop_process(task_id)) + self.assertNotIn(task_id, _user_stopped_tasks) + + +if __name__ == '__main__': + unittest.main()