From f4fb053e90c190230641a8e9edef4714608383c8 Mon Sep 17 00:00:00 2001 From: tandede <1090179959@qq.com> Date: Thu, 20 Aug 2026 08:13:56 +0800 Subject: [PATCH] Fix LiveCodeBench spawn worker compatibility --- .../datasets/livecodebench/evaluator.py | 26 ++++----- .../datasets/test_livecodebench_evaluator.py | 53 +++++++++++++++++-- 2 files changed, 63 insertions(+), 16 deletions(-) diff --git a/opencompass/datasets/livecodebench/evaluator.py b/opencompass/datasets/livecodebench/evaluator.py index 77e3e251c..6bf7c7e03 100644 --- a/opencompass/datasets/livecodebench/evaluator.py +++ b/opencompass/datasets/livecodebench/evaluator.py @@ -22,6 +22,18 @@ DEFAULT_MEMORY_LIMIT_BYTES = 4 * 1024 * 1024 * 1024 +def _run_test_in_subprocess(sample, generation, debug, result, metadata_list, + timeout, memory_limit_bytes): + from .testing_util import run_test + res, metadata = run_test(sample, + test=generation, + debug=debug, + timeout=timeout, + memory_limit_bytes=memory_limit_bytes) + result.append(res) + metadata_list.append(metadata) + + def codegen_check_correctness(sample, generation, timeout, @@ -32,23 +44,11 @@ def codegen_check_correctness(sample, The global timeout is to catch some extreme/rare cases not handled by the timeouts inside `run_test` """ - - def _temp_run(sample, generation, debug, result, metadata_list, timeout, - memory_limit_bytes): - from .testing_util import run_test - res, metadata = run_test(sample, - test=generation, - debug=debug, - timeout=timeout, - memory_limit_bytes=memory_limit_bytes) - result.append(res) - metadata_list.append(metadata) - manager = multiprocessing.Manager() result = manager.list() metadata_list = manager.list() p = multiprocessing.Process( - target=_temp_run, + target=_run_test_in_subprocess, args=(sample, generation, debug, result, metadata_list, timeout, memory_limit_bytes), ) diff --git a/tests/datasets/test_livecodebench_evaluator.py b/tests/datasets/test_livecodebench_evaluator.py index b22797115..512b0b662 100644 --- a/tests/datasets/test_livecodebench_evaluator.py +++ b/tests/datasets/test_livecodebench_evaluator.py @@ -1,11 +1,19 @@ import json +import multiprocessing import resource +import sys import unittest from unittest.mock import patch from opencompass.datasets.livecodebench import evaluator, testing_util +def _record_spawn_result(sample, generation, debug, result, metadata_list, + timeout, memory_limit_bytes): + result.append([True]) + metadata_list.append({'start_method': 'spawn'}) + + class TestLiveCodeBenchMemoryLimit(unittest.TestCase): def test_run_test_passes_memory_limit_to_reliability_guard(self): @@ -28,6 +36,8 @@ def test_run_test_passes_memory_limit_to_reliability_guard(self): mock_reliability_guard.assert_called_once_with( maximum_memory_bytes=123456) + @unittest.skipUnless('fork' in multiprocessing.get_all_start_methods(), + 'requires the fork start method') def test_codegen_check_correctness_passes_memory_limit_to_worker(self): def fake_run_test(sample, @@ -46,7 +56,9 @@ def fake_run_test(sample, }) } - with patch.object(testing_util, 'run_test', fake_run_test): + fork_process = multiprocessing.get_context('fork').Process + with patch.object(testing_util, 'run_test', fake_run_test), patch.object( + evaluator.multiprocessing, 'Process', fork_process): result, metadata = evaluator.codegen_check_correctness( sample, 'unused generation', @@ -57,6 +69,10 @@ def fake_run_test(sample, self.assertEqual(result, [True]) self.assertEqual(metadata['memory_limit_bytes'], 123456) + @unittest.skipUnless('fork' in multiprocessing.get_all_start_methods(), + 'requires the fork start method') + @unittest.skipIf(sys.platform == 'darwin', + 'macOS rejects limits below current virtual memory') def test_reliability_guard_sets_address_space_limit(self): child_memory_limit = 256 * 1024 * 1024 @@ -80,7 +96,9 @@ def fake_run_test(sample, }) } - with patch.object(testing_util, 'run_test', fake_run_test): + fork_process = multiprocessing.get_context('fork').Process + with patch.object(testing_util, 'run_test', fake_run_test), patch.object( + evaluator.multiprocessing, 'Process', fork_process): result, metadata = evaluator.codegen_check_correctness( sample, 'unused generation', @@ -92,6 +110,8 @@ def fake_run_test(sample, self.assertEqual(metadata['rlimit_as'], [child_memory_limit, child_memory_limit]) + @unittest.skipUnless('fork' in multiprocessing.get_all_start_methods(), + 'requires the fork start method') def test_codegen_check_correctness_returns_metadata_when_worker_exits( self): @@ -111,7 +131,9 @@ def fake_run_test(sample, }) } - with patch.object(testing_util, 'run_test', fake_run_test): + fork_process = multiprocessing.get_context('fork').Process + with patch.object(testing_util, 'run_test', fake_run_test), patch.object( + evaluator.multiprocessing, 'Process', fork_process): result, metadata = evaluator.codegen_check_correctness( sample, 'unused generation', @@ -123,6 +145,31 @@ def fake_run_test(sample, self.assertEqual(metadata['error_message'], 'Global Timeout or Memory Limit Exceeded') + def test_codegen_check_correctness_supports_spawn_workers(self): + sample = { + 'input_output': + json.dumps({ + 'inputs': ['1'], + 'outputs': ['1'], + 'fn_name': 'identity', + }) + } + + spawn_process = multiprocessing.get_context('spawn').Process + with patch.object(evaluator, '_run_test_in_subprocess', + _record_spawn_result), patch.object( + evaluator.multiprocessing, 'Process', + spawn_process): + result, metadata = evaluator.codegen_check_correctness( + sample, + 'unused generation', + timeout=20, + debug=False, + memory_limit_bytes=None) + + self.assertEqual(result, [True]) + self.assertEqual(metadata['start_method'], 'spawn') + if __name__ == '__main__': unittest.main()